TODO: Decouple this from Conversation and Message Manager so it can be unit tested.
160514
static PendingIntent generatePendingIntentFromApptentivePushData(Conversation conversation, String apptentivePushData) {
		ApptentiveLog.d(PUSH, "Generating Apptentive push PendingIntent.");
		if (!TextUtils.isEmpty(apptentivePushData)) {
			try {
				JSONObject pushJson = new JSONObject(apptentivePushData);

				// we need to check if current user is actually the receiver of this notification
				final String conversationId = pushJson.optString(PUSH_CONVERSATION_ID, null);
				if (conversationId != null) {
					// is it an actual receiver?
					if (!StringUtils.equal(conversation.getConversationId(), conversationId)) {
						ApptentiveLog.i(PUSH, "Can't generate pending intent from Apptentive push data: push conversation id doesn't match active conversation");
						return null;
					}
				}

				ApptentiveInternal.PushAction action = ApptentiveInternal.PushAction.unknown;
				if (pushJson.has(ApptentiveInternal.PUSH_ACTION)) {
					action = ApptentiveInternal.PushAction.parse(pushJson.getString(ApptentiveInternal.PUSH_ACTION));
				}
				switch (action) {
					case pmc: {
						// Prefetch message when push for message center is received
						MessageManager mgr = conversation.getMessageManager();
						if (mgr != null) {
							mgr.startMessagePreFetchTask();
						}
						// Construct a pending intent to launch message center
						return ApptentiveInternal.prepareMessageCenterPendingIntent(ApptentiveInternal.getInstance().getApplicationContext(), conversation);
					}
					default:
						ApptentiveLog.w(PUSH, "Unknown Apptentive push notification action: \"%s\"", action.name());
				}
			} catch (Exception e) {
				ApptentiveLog.e(PUSH, e, "Error parsing JSON from push notification.");
				logException(e);
			}
		}
		return null;
	}
1200835
public Observable<ComapiResult<ConversationDetails>> getConversation(@NonNull final String conversationId) {

        final String token = getToken();

        if (sessionController.isCreatingSession()) {
            return getTaskQueue().queueGetConversation(conversationId);
        } else if (TextUtils.isEmpty(token)) {
            return Observable.error(getSessionStateErrorDescription());
        } else {
            return doGetConversation(token, conversationId);
        }
    }
1158219
@SuppressWarnings("unchecked")
    protected void initMessageObject() {

        try {
            if (CmsStringUtil.isEmpty(getParamAction()) || CmsDialog.DIALOG_INITIAL.equals(getParamAction())) {
                // create a new list
                m_groups = new ArrayList<String>();
            } else {
                // this is not the initial call, get the message info object from session
                m_groups = (List<String>)getDialogObject();
            }
        } catch (Exception e) {
            // create a new list
            m_groups = new ArrayList<String>();
        }
    }
1180267
public void sendMessage(@NonNull final String conversationId, @NonNull final MessageToSend message, @Nullable Callback<ComapiResult<MessageSentResponse>> callback) {
        adapter.adapt(sendMessage(conversationId, message), callback);
    }
132494
private Chat createChat(Message message) {
        Jid from = message.getFrom();
        // According to RFC6120 8.1.2.1 4. messages without a 'from' attribute are valid, but they
        // are of no use in this case for ChatManager
        if (from == null) {
            return null;
        }

        EntityJid userJID = from.asEntityJidIfPossible();
        if (userJID == null) {
            LOGGER.warning("Message from JID without localpart: '" + message.toXML() + "'");
            return null;
        }
        String threadID = message.getThread();
        if (threadID == null) {
            threadID = nextID();
        }

        return createChat(userJID, threadID, false);
    }
1273761
private Map<String, Conversation> makeMapFromDownloadedConversations(List<Conversation> list) {

        Map<String, Conversation> map = new HashMap<>();

        if (list != null && !list.isEmpty()) {
            for (Conversation details : list) {
                map.put(details.getId(), details);
            }
        }

        return map;
    }
689774
public void free()
    {
        if (m_messageReceiver != null)
            m_messageReceiver.removeMessageFilter(this, false); // Remove from this and don't free again
        m_messageReceiver = null;
        
        while (this.getMessageListener(0) != null)
        {
            JMessageListener listener = this.getMessageListener(0);
            this.removeFilterMessageListener(listener);
        }
        if (m_vListenerList != null)
            m_vListenerList.clear();
        m_vListenerList = null;
        
        super.free();
        m_intID = null;
    }
1074876
Observable<ComapiResult<Void>> markDelivered(String conversationId, Set<String> ids) {

        final List<MessageStatusUpdate> updates = new ArrayList<>();
        updates.add(MessageStatusUpdate.builder().setMessagesIds(ids).setStatus(MessageStatus.delivered).setTimestamp(DateHelper.getCurrentUTC()).build());

        return checkState().flatMap(client -> client.service().messaging().updateMessageStatus(conversationId, updates)
                .retryWhen(observable -> {
                    return observable.zipWith(Observable.range(1, 3), (Func2<Throwable, Integer, Integer>) (throwable, integer) -> integer).flatMap(new Func1<Integer, Observable<Long>>() {
                        @Override
                        public Observable<Long> call(Integer retryCount) {
                            return Observable.timer((long) Math.pow(1, retryCount), TimeUnit.SECONDS);
                        }
                    });
                }
                ));
    }
445697
Observable<ChatResult> getPreviousMessages(final String conversationId) {

        return persistenceController.getConversation(conversationId)
                .map(conversation -> conversation != null ? conversation.getFirstLocalEventId() : null)
                .flatMap(from -> {

                    final Long queryFrom;

                    if (from != null) {

                        if (from == 0) {
                            return Observable.fromCallable(() -> new ChatResult(true, null));
                        } else if (from > 0) {
                            queryFrom = from - 1;
                        } else {
                            queryFrom = null;
                        }
                    } else {
                        queryFrom = null;
                    }

                    return checkState().flatMap(client -> client.service().messaging().queryMessages(conversationId, queryFrom, messagesPerQuery))
                            .flatMap(result -> persistenceController.processMessageQueryResponse(conversationId, result))
                            .flatMap(result -> persistenceController.processOrphanedEvents(result, orphanedEventsToRemoveListener))
                            .flatMap((Func1<ComapiResult<MessagesQueryResponse>, Observable<ChatResult>>) result -> (result.isSuccessful() && result.getResult().getMessages().isEmpty() && result.getResult().getEarliestEventId() > 0) ?
                                    getPreviousMessages(conversationId) :
                                    Observable.fromCallable(() -> new ChatResult(result.isSuccessful(), result.isSuccessful() ? null : new ChatResult.Error(result))));
                });
    }
130933
Observable<ChatResult> handleConversationCreated(ComapiResult<ConversationDetails> result) {
        if (result.isSuccessful()) {
            return persistenceController.upsertConversation(ChatConversation.builder().populate(result.getResult(), result.getETag()).build()).map(success -> adapter.adaptResult(result, success));
        } else {
            return Observable.fromCallable(() -> adapter.adaptResult(result));
        }
    }
745804
private synchronized void saveConversationData() throws SerializerException {
		if (ApptentiveLog.canLog(ApptentiveLog.Level.VERBOSE)) {
			ApptentiveLog.v(CONVERSATION, "Saving conversation data...");
			ApptentiveLog.v(CONVERSATION, "EventData: %s", getEventData().toString());
			ApptentiveLog.v(CONVERSATION, "Messages: %s", messageManager.getMessageStore().toString());
		}
		long start = System.currentTimeMillis();

		FileSerializer serializer = new EncryptedFileSerializer(conversationDataFile, encryptionKey);
		serializer.serialize(conversationData);
		ApptentiveLog.v(CONVERSATION, "Conversation data saved (took %d ms)", System.currentTimeMillis() - start);
	}
1052280
private Observable<Boolean> upsertTempMessage(ChatMessage message) {
        return persistenceController.updateStoreWithNewMessage(message, noConversationListener)
                .doOnError(t -> log.e("Error saving temp message " + t.getLocalizedMessage()))
                .onErrorReturn(t -> false);
    }
558689
public Observable<ComapiResult<MessageSentResponse>> sendMessage(@NonNull final String conversationId, @NonNull final MessageToSend message) {

        final String token = getToken();

        if (sessionController.isCreatingSession()) {
            return getTaskQueue().queueSendMessage(conversationId, message);
        } else if (TextUtils.isEmpty(token)) {
            return Observable.error(getSessionStateErrorDescription());
        } else {
            return doSendMessage(token, conversationId, message);
        }
    }
306523
Observable<ChatResult> handleNoLocalConversation(String conversationId) {

        return checkState().flatMap(client -> client.service().messaging().getConversation(conversationId)
                .flatMap(result -> {
                    if (result.isSuccessful() && result.getResult() != null) {
                        return persistenceController.upsertConversation(ChatConversation.builder().populate(result.getResult(), result.getETag()).build())
                                .map(success -> new ChatResult(success, success ? null : new ChatResult.Error(0, "External store reported failure.", "Error when inserting conversation "+conversationId)));
                    } else {
                        return Observable.fromCallable(() -> adapter.adaptResult(result));
                    }
                }));
    }
501210
private ResourceBundleHolder loadMessageBundle(Method m) throws ClassNotFoundException {
        Class c = m.getDeclaringClass();
        Class<?> msg = c.getClassLoader().loadClass(c.getName().substring(0, c.getName().lastIndexOf(".")) + ".Messages");
        return ResourceBundleHolder.get(msg);
    }
790532
protected void processReceivedMessage (ChatMessage msg, String localtype)
    {
        String autoResponse = null;
        Name speaker = null;
        Name speakerDisplay = null;
        byte mode = (byte)-1;

        // figure out if the message was triggered by another user
        if (msg instanceof UserMessage) {
            UserMessage umsg = (UserMessage)msg;
            speaker = umsg.speaker;
            speakerDisplay = umsg.getSpeakerDisplayName();
            mode = umsg.mode;

        } else if (msg instanceof UserSystemMessage) {
            speaker = ((UserSystemMessage)msg).speaker;
            speakerDisplay = speaker;
        }

        // Translate and timestamp the message. This would happen during dispatch but we
        // need to do it ahead of filtering.
        setClientInfo(msg, localtype);

        // if there was an originating speaker, see if we want to hear it
        if (speaker != null) {
            if (shouldFilter(msg) && (msg.message = filter(msg.message, speaker, false)) == null) {
                return;
            }

            if (USER_CHAT_TYPE.equals(localtype) &&
                mode == ChatCodes.DEFAULT_MODE) {
                // if it was a tell, add the speaker as a chatter
                addChatter(speaker);

                // note whether or not we have an auto-response
                BodyObject self = (BodyObject)_ctx.getClient().getClientObject();
                if (!StringUtil.isBlank(self.awayMessage)) {
                    autoResponse = self.awayMessage;
                }
            }
        }

        // and send it off!
        dispatchMessage(msg, localtype);

        // if we auto-responded, report as much
        if (autoResponse != null) {
            String amsg = MessageBundle.tcompose(
                "m.auto_responded", speakerDisplay, autoResponse);
            displayFeedback(_bundle, amsg);
        }
    }
807273
Observable<ChatConversationBase> getConversation(@NonNull String conversationId) {

        return Observable.create(emitter -> storeFactory.execute(new StoreTransaction<ChatStore>() {
            @Override
            protected void execute(ChatStore store) {
                store.open();
                ChatConversationBase c = store.getConversation(conversationId);
                store.close();
                emitter.onNext(c);
                emitter.onCompleted();
            }
        }), Emitter.BackpressureMode.BUFFER);
    }