contents;
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/model/AbstractMessagePayloadModel.java b/data/src/main/java/org/mercury_im/messenger/data/model/AbstractMessagePayloadModel.java
new file mode 100644
index 0000000..00a7d76
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/model/AbstractMessagePayloadModel.java
@@ -0,0 +1,27 @@
+package org.mercury_im.messenger.data.model;
+
+import org.mercury_im.messenger.data.enums.MessageContentType;
+
+import io.requery.Column;
+import io.requery.Entity;
+import io.requery.Generated;
+import io.requery.Key;
+import io.requery.ManyToOne;
+import io.requery.Table;
+
+@Entity
+@Table(name = "msg_payloads")
+public abstract class AbstractMessagePayloadModel {
+
+ @Key @Generated
+ long id;
+
+ @ManyToOne
+ MessagePayloadContainerModel payloadContainer;
+
+ @Column
+ String body;
+
+ @Column(nullable = false)
+ MessageContentType type;
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/model/AbstractPeerModel.java b/data/src/main/java/org/mercury_im/messenger/data/model/AbstractPeerModel.java
new file mode 100644
index 0000000..5a4e13e
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/model/AbstractPeerModel.java
@@ -0,0 +1,56 @@
+package org.mercury_im.messenger.data.model;
+
+import org.mercury_im.messenger.data.converter.SubscriptionDirectionConverter;
+import org.mercury_im.messenger.entity.contact.SubscriptionDirection;
+
+import java.util.UUID;
+
+import io.requery.CascadeAction;
+import io.requery.Column;
+import io.requery.Convert;
+import io.requery.Entity;
+import io.requery.ForeignKey;
+import io.requery.Generated;
+import io.requery.Index;
+import io.requery.Key;
+import io.requery.ManyToOne;
+import io.requery.Persistable;
+import io.requery.ReferentialAction;
+import io.requery.Table;
+import io.requery.converter.UUIDConverter;
+
+@Entity
+@Table(name = "contacts", uniqueIndexes = "unique_address")
+public abstract class AbstractPeerModel implements Persistable {
+
+ @Key
+ @Convert(UUIDConverter.class)
+ UUID id;
+
+ @Index("unique_address")
+ @ManyToOne(cascade = CascadeAction.NONE)
+ @ForeignKey(referencedColumn = "id")
+ AccountModel account;
+
+ @Index("unique_address")
+ @Column(nullable = false)
+ String address;
+
+ @Column
+ String name;
+
+ @Convert(SubscriptionDirectionConverter.class)
+ SubscriptionDirection subscriptionDirection;
+
+ boolean subscriptionPending;
+
+ boolean subscriptionPreApproved;
+
+ @Override
+ public String toString() {
+ return "Peer[" + id + ", " +
+ name + ", " +
+ address + ", " +
+ account + "]";
+ }
+}
diff --git a/persistence/src/main/java/org/mercury_im/messenger/persistence/entity/AbstractSaslAuthenticationResultModel.java b/data/src/main/java/org/mercury_im/messenger/data/model/AbstractSaslAuthenticationResultModel.java
similarity index 75%
rename from persistence/src/main/java/org/mercury_im/messenger/persistence/entity/AbstractSaslAuthenticationResultModel.java
rename to data/src/main/java/org/mercury_im/messenger/data/model/AbstractSaslAuthenticationResultModel.java
index 1753071..f13b416 100644
--- a/persistence/src/main/java/org/mercury_im/messenger/persistence/entity/AbstractSaslAuthenticationResultModel.java
+++ b/data/src/main/java/org/mercury_im/messenger/data/model/AbstractSaslAuthenticationResultModel.java
@@ -1,6 +1,6 @@
-package org.mercury_im.messenger.persistence.entity;
+package org.mercury_im.messenger.data.model;
-import org.mercury_im.messenger.persistence.enums.SaslCondition;
+import org.mercury_im.messenger.data.enums.SaslCondition;
import io.requery.Entity;
import io.requery.Key;
diff --git a/data/src/main/java/org/mercury_im/messenger/data/model/package-info.java b/data/src/main/java/org/mercury_im/messenger/data/model/package-info.java
new file mode 100644
index 0000000..77cc9dd
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/model/package-info.java
@@ -0,0 +1,12 @@
+/**
+ * The model package contains requery model definitions.
+ * All of those classes are abstract, since the requery framework generates concrete implementations
+ * during compilation. Those files can later be found in
+ * build/generated/sources/annotationProcessor/java/main/...
.
+ *
+ * The structure of the model classes closely mimics the structure of their entity pendants
+ * declared in the entity
module.
+ *
+ * @see requery wiki on model definitions
+ */
+package org.mercury_im.messenger.data.model;
diff --git a/persistence/src/main/java/org/mercury_im/messenger/persistence/repository/RequeryRepository.java b/data/src/main/java/org/mercury_im/messenger/data/repository/RequeryRepository.java
similarity index 70%
rename from persistence/src/main/java/org/mercury_im/messenger/persistence/repository/RequeryRepository.java
rename to data/src/main/java/org/mercury_im/messenger/data/repository/RequeryRepository.java
index 4cf203d..8b814b2 100644
--- a/persistence/src/main/java/org/mercury_im/messenger/persistence/repository/RequeryRepository.java
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/RequeryRepository.java
@@ -1,4 +1,8 @@
-package org.mercury_im.messenger.persistence.repository;
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import javax.inject.Named;
import io.reactivex.Scheduler;
import io.requery.Persistable;
@@ -12,8 +16,8 @@ public abstract class RequeryRepository {
private final ReactiveEntityStore data;
protected RequeryRepository(ReactiveEntityStore data,
- Scheduler subscriberScheduler,
- Scheduler observerScheduler) {
+ @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler) {
this.data = data;
this.subscriberScheduler = subscriberScheduler;
this.observerScheduler = observerScheduler;
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/XmppAccountRepository.java b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppAccountRepository.java
new file mode 100644
index 0000000..f683023
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppAccountRepository.java
@@ -0,0 +1,146 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.data.mapping.AccountMapping;
+import org.mercury_im.messenger.data.model.AccountModel;
+import org.mercury_im.messenger.data.repository.dao.AccountDao;
+import org.mercury_im.messenger.util.Optional;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Scheduler;
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.query.ResultDelegate;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveResult;
+
+public class XmppAccountRepository
+ extends RequeryRepository
+ implements AccountRepository {
+
+ private final AccountMapping accountMapping;
+ private final AccountDao dao;
+
+ @Inject
+ public XmppAccountRepository(ReactiveEntityStore data,
+ @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler,
+ AccountMapping accountMapping) {
+ super(data, subscriberScheduler, observerScheduler);
+ this.accountMapping = accountMapping;
+ this.dao = new AccountDao(data);
+ }
+
+ @Override
+ public Single insertAccount(Account account) {
+ return Single.just(account)
+ .map(accountMapping::toModel)
+ .flatMap(dao::insert)
+ .map(model -> accountMapping.toEntity(model, account))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeAccount(UUID accountId) {
+ return dao.get(accountId).observableResult()
+ .map(result -> new Optional<>(result.firstOrNull()))
+ .map(accountMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe getAccount(UUID accountId) {
+ return dao.get(accountId).maybe()
+ .map(accountMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeAccountByAddress(String address) {
+ return dao.get(address).observableResult()
+ .map(result -> new Optional<>(result.firstOrNull()))
+ .map(accountMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe getAccountByAddress(String address) {
+ return dao.get(address).maybe()
+ .map(accountMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeAllAccounts() {
+ return dao.getAll().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::modelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable observeAccounts() {
+ return dao.getAll().observableResult()
+ .flatMap(ReactiveResult::observable)
+ .map(accountMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single updateAccount(Account account) {
+ // Since we cannot access setId() of AccountModel, we have to query the model by ID and update it manually.
+ // https://github.com/requery/requery/issues/616#issuecomment-315685460
+
+ return dao.get(account.getId()).maybe().toSingle()
+ .map(model -> accountMapping.toModel(account, model))
+ .flatMap(updatedModel -> data().update(updatedModel))
+ .map(model -> accountMapping.toEntity(model, account))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single upsertAccount(Account account) {
+ return dao.get(account.getId()).maybe()
+ .switchIfEmpty(
+ Single.just(account).map(accountMapping::toModel).flatMap(dao::insert))
+ .map(model -> accountMapping.toModel(account, model))
+ .flatMap(data()::update)
+ .map(model -> accountMapping.toEntity(model, account))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Completable deleteAccount(UUID accountId) {
+ return dao.delete(accountId).ignoreElement()
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ private List modelsToEntities(List models) {
+ List entities = new ArrayList<>(models.size());
+ for (AccountModel model : models) {
+ entities.add(accountMapping.toEntity(model));
+ }
+ return entities;
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/XmppDirectChatRepository.java b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppDirectChatRepository.java
new file mode 100644
index 0000000..08462a4
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppDirectChatRepository.java
@@ -0,0 +1,153 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.data.mapping.DirectChatMapping;
+import org.mercury_im.messenger.data.model.DirectChatModel;
+import org.mercury_im.messenger.data.repository.dao.DirectChatDao;
+import org.mercury_im.messenger.util.Optional;
+import org.mercury_im.messenger.entity.chat.DirectChat;
+import org.mercury_im.messenger.entity.chat.IDirectChat;
+import org.mercury_im.messenger.entity.contact.Peer;
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Scheduler;
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.query.ResultDelegate;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveResult;
+
+public class XmppDirectChatRepository
+ extends RequeryRepository
+ implements DirectChatRepository {
+
+ private final DirectChatMapping directChatMapping;
+
+ private final DirectChatDao dao;
+
+ @Inject
+ public XmppDirectChatRepository(
+ ReactiveEntityStore data,
+ @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler,
+ DirectChatMapping directChatMapping) {
+ super(data, subscriberScheduler, observerScheduler);
+ this.directChatMapping = directChatMapping;
+ this.dao = new DirectChatDao(data);
+ }
+
+ @Override
+ public Single insertDirectChat(DirectChat chat) {
+ return Single.just(chat)
+ // map entity to model
+ .map(directChatMapping::toModel)
+ .flatMap(dao::insert)
+ // map back to entity
+ .map(model -> directChatMapping.toEntity(model, chat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeDirectChat(UUID chatId) {
+ return dao.get(chatId).observableResult()
+ .map(result -> new Optional<>(result.firstOrNull()))
+ .map(directChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe getDirectChat(UUID chatId) {
+ return dao.get(chatId).maybe()
+ .map(directChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single getOrCreateChatWithPeer(Peer peer) {
+ return getDirectChatByPeer(peer)
+ .switchIfEmpty(Single.just(new IDirectChat())
+ .map(chat -> {
+ chat.setAccount(peer.getAccount());
+ chat.setPeer(peer);
+ return chat;
+ })
+ .flatMap(this::insertDirectChat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeDirectChatByPeer(Peer peer) {
+ return dao.getByPeer(peer.getId()).observableResult()
+ .map(result -> new Optional<>(result.firstOrNull()))
+ .map(directChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe getDirectChatByPeer(Peer peer) {
+ return dao.getByPeer(peer.getId()).maybe()
+ .map(directChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeAllDirectChats() {
+ return dao.getAll().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::chatModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ private List chatModelsToEntities(List models) {
+ List entities = new ArrayList<>(models.size());
+ for (DirectChatModel model : models) {
+ entities.add(directChatMapping.toEntity(model));
+ }
+ return entities;
+ }
+
+ @Override
+ public Single updateDirectChat(DirectChat chat) {
+ return dao.get(chat.getId()).maybe().toSingle()
+ .map(model -> directChatMapping.toModel(chat, model))
+ .flatMap(data()::update)
+ .map(model -> directChatMapping.toEntity(model, chat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single upsertDirectChat(DirectChat chat) {
+ return dao.get(chat.getId()).maybe()
+ .switchIfEmpty(dao.insert(directChatMapping.toModel(chat)))
+ .map(directChatModel -> directChatMapping.toModel(chat, directChatModel))
+ .flatMap(data()::update)
+ .map(model -> directChatMapping.toEntity(model, chat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Completable deleteDirectChat(UUID chatId) {
+ return dao.delete(chatId)
+ .ignoreElement()
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/XmppEntityCapsRepository.java b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppEntityCapsRepository.java
new file mode 100644
index 0000000..ce9d4cf
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppEntityCapsRepository.java
@@ -0,0 +1,91 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.data.mapping.EntityCapsMapping;
+import org.mercury_im.messenger.data.model.EntityCapsModel;
+import org.mercury_im.messenger.entity.caps.EntityCapsRecord;
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Scheduler;
+import io.requery.Persistable;
+import io.requery.query.Expression;
+import io.requery.query.ResultDelegate;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveResult;
+
+public class XmppEntityCapsRepository extends RequeryRepository implements EntityCapsRepository {
+
+ private final EntityCapsMapping entityCapsMapping;
+
+ @Inject
+ public XmppEntityCapsRepository(
+ ReactiveEntityStore data,
+ @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler,
+ EntityCapsMapping mapping) {
+ super(data, subscriberScheduler, observerScheduler);
+ this.entityCapsMapping = mapping;
+ }
+
+ @Override
+ public Observable> observeAllEntityCapsRecords() {
+ return data().select(EntityCapsModel.class).get()
+ .observableResult()
+ .map(result -> result.toMap(EntityCapsModel.NODE_VER, new ConcurrentHashMap<>()))
+ .map(this::mapModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ private Map mapModelsToEntities(Map models) {
+ Map entities = new ConcurrentHashMap<>();
+ for (String key : models.keySet()) {
+ entities.put(key, entityCapsMapping.toEntity(models.get(key)));
+ }
+ return entities;
+ }
+
+ private Map mapEntitiesToModels(Map entities) {
+ Map models = new ConcurrentHashMap<>();
+ for (String key : entities.keySet()) {
+ models.put(key, entityCapsMapping.toModel(entities.get(key)));
+ }
+ return models;
+ }
+
+ @Override
+ public Observable observeEntityCapsRecords() {
+ return data().select(EntityCapsModel.class)
+ .get().observableResult()
+ .flatMap(ReactiveResult::observable)
+ .map(entityCapsMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe maybeGetEntityCapsRecord(String nodeVer) {
+ return data().select(EntityCapsModel.class)
+ .where(EntityCapsModel.NODE_VER.eq(nodeVer))
+ .get().maybe()
+ .map(entityCapsMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Completable insertEntityCapsRecord(EntityCapsRecord entityCapsRecord) {
+ return data().upsert(entityCapsMapping.toModel(entityCapsRecord))
+ .ignoreElement()
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/XmppGroupChatRepository.java b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppGroupChatRepository.java
new file mode 100644
index 0000000..d43c316
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppGroupChatRepository.java
@@ -0,0 +1,161 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.data.mapping.GroupChatMapping;
+import org.mercury_im.messenger.data.model.GroupChatModel;
+import org.mercury_im.messenger.data.repository.dao.GroupChatDao;
+import org.mercury_im.messenger.util.Optional;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.chat.GroupChat;
+import org.mercury_im.messenger.entity.chat.IGroupChat;
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Scheduler;
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.query.ResultDelegate;
+import io.requery.reactivex.ReactiveEntityStore;
+
+public class XmppGroupChatRepository
+ extends RequeryRepository
+ implements GroupChatRepository {
+
+ private final GroupChatMapping groupChatMapping;
+
+ private final GroupChatDao dao;
+
+ @Inject
+ public XmppGroupChatRepository(
+ ReactiveEntityStore data,
+ @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler,
+ GroupChatMapping groupChatMapping) {
+ super(data, subscriberScheduler, observerScheduler);
+ this.groupChatMapping = groupChatMapping;
+ this.dao = new GroupChatDao(data);
+ }
+
+ @Override
+ public Single insertGroupChat(GroupChat chat) {
+ return Single.just(chat)
+ .map(groupChatMapping::toModel)
+ .flatMap(dao::insert)
+ .map(model -> groupChatMapping.toEntity(model, chat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeGroupChat(UUID chatId) {
+ return dao.get(chatId).observableResult()
+ .map(result -> new Optional<>(result.firstOrNull()))
+ .map(groupChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe getGroupChat(UUID chatId) {
+ return dao.get(chatId).maybe()
+ .map(groupChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single getOrCreateGroupChat(Account account, String roomAddress) {
+ return getGroupChatByRoomAddress(account, roomAddress)
+ .switchIfEmpty(Single.just(new IGroupChat())
+ .map(chat -> {
+ chat.setAccount(account);
+ chat.setRoomAddress(roomAddress);
+ return chat;
+ })
+ .flatMap(this::insertGroupChat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeGroupChatByRoomAddress(UUID accountId, String roomAddress) {
+ return dao.get(accountId, roomAddress).observableResult()
+ .map(result -> new Optional<>(result.firstOrNull()))
+ .map(groupChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe getGroupChatByRoomAddress(UUID accountId, String roomAddress) {
+ return dao.get(accountId, roomAddress).maybe()
+ .map(groupChatMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeAllGroupChats() {
+ return dao.getAll().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::modelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeAllGroupChatsOfAccount(UUID accountId) {
+ return data().select(GroupChatModel.class)
+ .where(GroupChatModel.ACCOUNT_ID.eq(accountId))
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::modelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ private List modelsToEntities(List models) {
+ List entities = new ArrayList<>(models.size());
+ for (GroupChatModel model : models) {
+ entities.add(groupChatMapping.toEntity(model));
+ }
+ return entities;
+ }
+
+ @Override
+ public Single updateGroupChat(GroupChat chat) {
+ return dao.get(chat.getId()).maybe().toSingle()
+ .map(model -> groupChatMapping.toModel(chat, model))
+ .flatMap(data()::update)
+ .map(model -> groupChatMapping.toEntity(model, chat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single upsertGroupChat(GroupChat chat) {
+ return dao.get(chat.getId()).maybe()
+ .switchIfEmpty(Single.just(chat).map(groupChatMapping::toModel)
+ .flatMap(dao::insert))
+ .map(model -> groupChatMapping.toModel(chat, model))
+ .flatMap(data()::update)
+ .map(model -> groupChatMapping.toEntity(model, chat))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Completable deleteGroupChat(UUID chatId) {
+ return dao.delete(chatId).ignoreElement()
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/XmppMessageRepository.java b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppMessageRepository.java
new file mode 100644
index 0000000..9ef7259
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppMessageRepository.java
@@ -0,0 +1,231 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.data.mapping.DirectChatMapping;
+import org.mercury_im.messenger.data.mapping.GroupChatMapping;
+import org.mercury_im.messenger.data.mapping.MessageMapping;
+import org.mercury_im.messenger.data.model.DirectChatModel;
+import org.mercury_im.messenger.data.model.DirectMessagesRelation;
+import org.mercury_im.messenger.data.model.GroupChatModel;
+import org.mercury_im.messenger.data.model.GroupMessagesRelation;
+import org.mercury_im.messenger.data.model.MessageModel;
+import org.mercury_im.messenger.data.model.MessagePayloadContainerModel;
+import org.mercury_im.messenger.data.model.MessagePayloadModel;
+import org.mercury_im.messenger.data.repository.dao.DirectChatDao;
+import org.mercury_im.messenger.data.repository.dao.GroupChatDao;
+import org.mercury_im.messenger.data.repository.dao.MessageDao;
+import org.mercury_im.messenger.entity.chat.DirectChat;
+import org.mercury_im.messenger.entity.chat.GroupChat;
+import org.mercury_im.messenger.entity.message.Message;
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+
+import io.reactivex.Completable;
+import io.reactivex.Observable;
+import io.reactivex.Scheduler;
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.query.ResultDelegate;
+import io.requery.reactivex.ReactiveEntityStore;
+
+public class XmppMessageRepository
+ extends RequeryRepository
+ implements MessageRepository {
+
+ private final MessageMapping messageMapping;
+ private final DirectChatMapping directChatMapping;
+ private final GroupChatMapping groupChatMapping;
+
+ private final DirectChatDao directChatDao;
+ private final GroupChatDao groupChatDao;
+
+ private final MessageDao dao;
+
+
+ @Inject
+ public XmppMessageRepository(ReactiveEntityStore data,
+ @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler,
+ MessageMapping messageMapping,
+ DirectChatMapping directChatMapping,
+ GroupChatMapping groupChatMapping) {
+ super(data, subscriberScheduler, observerScheduler);
+ this.messageMapping = messageMapping;
+ this.directChatMapping = directChatMapping;
+ this.groupChatMapping = groupChatMapping;
+ this.directChatDao = new DirectChatDao(data);
+ this.groupChatDao = new GroupChatDao(data);
+ this.dao = new MessageDao(data);
+ }
+
+ @Override
+ public Single insertMessage(DirectChat chat, Message message) {
+ return directChatDao.get(chat.getId()).maybe()
+ .switchIfEmpty(directChatDao.insert(directChatMapping.toModel(chat)))
+ .map(chatModel -> toRelation(chatModel, messageMapping.toModel(message)))
+ .flatMap(data()::insert)
+ .map(DirectMessagesRelation::getMessage)
+ .map(messageModel -> messageMapping.toEntity(messageModel, message))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single insertMessage(GroupChat chat, Message message) {
+ return groupChatDao.get(chat.getId()).maybe()
+ .switchIfEmpty(groupChatDao.insert(groupChatMapping.toModel(chat)))
+ .map(chatModel -> toRelation(chatModel, messageMapping.toModel(message)))
+ .flatMap(data()::insert)
+ .map(GroupMessagesRelation::getMessage)
+ .map(messageModel -> messageMapping.toEntity(messageModel, message))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeMessages(DirectChat chat) {
+ return data().select(MessageModel.class)
+ .from(MessageModel.class)
+ .join(DirectMessagesRelation.class)
+ .on(DirectMessagesRelation.MESSAGE_ID.eq(MessageModel.ID))
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::messageModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeMessages(GroupChat chat) {
+ return data().select(MessageModel.class)
+ .from(MessageModel.class)
+ .join(GroupMessagesRelation.class)
+ .on(GroupMessagesRelation.MESSAGE_ID.eq(MessageModel.ID))
+ .where(GroupMessagesRelation.CHAT_ID.eq(chat.getId()))
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::messageModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> findMessagesWithBody(String body) {
+ return data().select(MessageModel.class)
+ .from(MessageModel.class)
+ .join(MessagePayloadContainerModel.class)
+ .on(MessagePayloadContainerModel.MESSAGE_ID.eq(MessageModel.ID))
+ .join(MessagePayloadModel.class)
+ .on(MessagePayloadModel.PAYLOAD_CONTAINER_ID.eq(MessagePayloadContainerModel.ID))
+ .where(MessagePayloadModel.BODY.eq(body))
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::messageModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> findMessagesWithBody(DirectChat chat, String body) {
+ return data().select(MessageModel.class)
+
+ .from(MessageModel.class)
+ .join(DirectMessagesRelation.class)
+ .on(DirectMessagesRelation.MESSAGE_ID.eq(MessageModel.ID))
+ .join(MessagePayloadContainerModel.class)
+ .on(MessagePayloadContainerModel.MESSAGE_ID.eq(MessageModel.ID))
+ .join(MessagePayloadModel.class)
+ .on(MessagePayloadModel.PAYLOAD_CONTAINER_ID.eq(MessagePayloadContainerModel.ID))
+
+ .where(MessagePayloadModel.BODY.eq(body))
+ .and(DirectMessagesRelation.CHAT_ID.eq(chat.getId()))
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::messageModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> findMessagesWithBody(GroupChat chat, String body) {
+ return data().select(MessageModel.class)
+
+ .from(MessageModel.class)
+ .join(GroupMessagesRelation.class)
+ .on(GroupMessagesRelation.MESSAGE_ID.eq(MessageModel.ID))
+ .join(MessagePayloadContainerModel.class)
+ .on(MessagePayloadContainerModel.MESSAGE_ID.eq(MessageModel.ID))
+ .join(MessagePayloadModel.class)
+ .on(MessagePayloadModel.PAYLOAD_CONTAINER_ID.eq(MessagePayloadContainerModel.ID))
+
+ .where(MessagePayloadModel.BODY.eq(body))
+ .and(GroupMessagesRelation.CHAT_ID.eq(chat.getId()))
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::messageModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single upsertMessage(DirectChat chat, Message message) {
+ return null;
+ /*
+ return data().select(DirectMessagesRelation.class)
+ .where(DirectMessagesRelation.CHAT_ID.eq(chat.getId()))
+ .and(DirectMessagesRelation.MESSAGE_ID.eq(message.getId()))
+ .get().maybe()
+ .switchIfEmpty(Single.just(message)
+ .map(messageMapping::toEntity)
+ .flatMap(messageModel -> )
+ */
+ }
+
+ @Override
+ public Single upsertMessage(GroupChat chat, Message message) {
+ return null;
+ }
+
+ @Override
+ public Single updateMessage(Message message) {
+ return dao.get(message.getId()).maybe().toSingle()
+ .map(model -> messageMapping.toModel(message, model))
+ .flatMap(data()::update)
+ .map(model -> messageMapping.toEntity(model, message))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Completable deleteMessage(Message message) {
+ return data().delete(MessageModel.class)
+ .where(MessageModel.ID.eq(message.getId()))
+ .get().single().ignoreElement();
+ }
+
+ private DirectMessagesRelation toRelation(DirectChatModel chat, MessageModel message) {
+ DirectMessagesRelation relation = new DirectMessagesRelation();
+ relation.setChat(chat);
+ relation.setMessage(message);
+ return relation;
+ }
+
+ private GroupMessagesRelation toRelation(GroupChatModel chat, MessageModel message) {
+ GroupMessagesRelation relation = new GroupMessagesRelation();
+ relation.setChat(chat);
+ relation.setMessage(message);
+ return relation;
+ }
+
+ private List messageModelsToEntities(List models) {
+ List entities = new ArrayList<>(models.size());
+ for (MessageModel model : models) {
+ entities.add(messageMapping.toEntity(model));
+ }
+ return entities;
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/XmppPeerRepository.java b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppPeerRepository.java
new file mode 100644
index 0000000..b4194d4
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/XmppPeerRepository.java
@@ -0,0 +1,208 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.data.mapping.PeerMapping;
+import org.mercury_im.messenger.data.model.PeerModel;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.contact.IPeer;
+import org.mercury_im.messenger.entity.contact.Peer;
+import org.mercury_im.messenger.entity.contact.SubscriptionDirection;
+import org.mercury_im.messenger.util.Optional;
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Scheduler;
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.query.Expression;
+import io.requery.query.LogicalCondition;
+import io.requery.query.ResultDelegate;
+import io.requery.reactivex.ReactiveEntityStore;
+
+public class XmppPeerRepository
+ extends RequeryRepository
+ implements PeerRepository {
+
+ private final AccountRepository accountRepository;
+
+ private final PeerMapping peerMapping;
+
+ @Inject
+ public XmppPeerRepository(ReactiveEntityStore data,
+ @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler,
+ PeerMapping peerMapping, AccountRepository accountRepository) {
+ super(data, subscriberScheduler, observerScheduler);
+ this.peerMapping = peerMapping;
+ this.accountRepository = accountRepository;
+ }
+
+ @Override
+ public Single insertPeer(Peer peer) {
+ return data().insert(peerMapping.toModel(peer, new PeerModel()))
+ .map(model -> peerMapping.toEntity(model, peer))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observePeer(UUID peerId) {
+ return data().select(PeerModel.class)
+ .where(PeerModel.ID.eq(peerId))
+ .get().observableResult()
+ .map(result -> new Optional<>(peerMapping.toEntity(result.firstOrNull(), new IPeer())))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Maybe getPeer(UUID peerId) {
+ return data().select(PeerModel.class)
+ .where(PeerModel.ID.eq(peerId))
+ .get().maybe()
+ .map(model -> peerMapping.toEntity(model, new IPeer()))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observePeerByAddress(UUID accountId, String address) {
+ return data().select(PeerModel.class)
+ .where(PeerModel.ACCOUNT_ID.eq(accountId))
+ .and(PeerModel.ADDRESS.eq(address))
+ .get().observableResult()
+ .map(result -> new Optional<>(result.firstOrNull()))
+ .map(peerMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single getOrCreatePeer(UUID accountId, String address) {
+ return accountRepository.getAccount(accountId).toSingle()
+ .flatMap(account -> getOrCreatePeer(account, address));
+ }
+
+ @Override
+ public Maybe getPeerByAddress(UUID accountId, String address) {
+ return data().select(PeerModel.class)
+ .where(PeerModel.ACCOUNT_ID.eq(accountId))
+ .and(PeerModel.ADDRESS.eq(address))
+ .get().maybe()
+ .map(peerMapping::toEntity)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single getOrCreatePeer(Account account, String address) {
+ return getPeerByAddress(account, address)
+ .switchIfEmpty(Single
+ .just(new IPeer())
+ .map(peer -> {
+ peer.setAccount(account);
+ peer.setAddress(address);
+ return peer;
+ })
+ .flatMap(this::insertPeer))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Observable> observeAllPeers() {
+ return data().select(PeerModel.class)
+ .orderBy(PeerModel.ADDRESS)
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::peerModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ private List peerModelsToEntities(List peerModels) {
+ List peerEntities = new ArrayList<>(peerModels.size());
+ for (PeerModel model : peerModels) {
+ peerEntities.add(peerMapping.toEntity(model));
+ }
+ return peerEntities;
+ }
+
+ @Override
+ public Observable> observeAllContactsOfAccount(UUID accountId) {
+ return data().select(PeerModel.class)
+ .where(PeerModel.ACCOUNT_ID.eq(accountId))
+ .and(isContact())
+ .get().observableResult()
+ .map(ResultDelegate::toList)
+ .map(this::peerModelsToEntities)
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ private LogicalCondition, ?> isContact() {
+ return PeerModel.SUBSCRIPTION_DIRECTION.in(Arrays.asList(
+ SubscriptionDirection.both,
+ SubscriptionDirection.to));
+ }
+
+ @Override
+ public Single updatePeer(Peer peer) {
+ // In order to update, we fetch the model, update it and write it back.
+ return data().select(PeerModel.class)
+ .where(PeerModel.ID.eq(peer.getId()))
+ .get().maybe().toSingle()
+ // write changes into model
+ .map(model -> peerMapping.toModel(peer, model))
+ .flatMap(data()::update)
+ .map(model -> peerMapping.toEntity(model, peer))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Single upsertPeer(Peer peer) {
+ return data().select(PeerModel.class)
+ .where(PeerModel.ID.eq(peer.getId()))
+ .get().maybe()
+ // if not exists, create
+ .switchIfEmpty(Single.just(peer)
+ .map(peerMapping::toModel)
+ .flatMap(data()::insert))
+ // write changes into fetched model
+ .map(model -> peerMapping.toModel(peer, model))
+ // write changed model back to db
+ .flatMap(data()::update)
+ .map(model -> peerMapping.toEntity(model, peer))
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Completable deletePeer(Peer peer) {
+ return data().delete(PeerModel.class)
+ .where(PeerModel.ID.eq(peer.getId()))
+ .get().single().ignoreElement()
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+
+ @Override
+ public Completable deletePeer(UUID accountId, String address) {
+ return data().delete(PeerModel.class)
+ .where(PeerModel.ACCOUNT_ID.eq(accountId)
+ .and(PeerModel.ADDRESS.eq(address)))
+ .get().single().ignoreElement()
+ .subscribeOn(subscriberScheduler())
+ .observeOn(observerScheduler());
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/dao/AccountDao.java b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/AccountDao.java
new file mode 100644
index 0000000..3aaf9d4
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/AccountDao.java
@@ -0,0 +1,47 @@
+package org.mercury_im.messenger.data.repository.dao;
+
+import org.mercury_im.messenger.data.model.AccountModel;
+
+import java.util.UUID;
+
+import javax.inject.Inject;
+
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveResult;
+
+public class AccountDao extends RequeryDao {
+
+ @Inject
+ public AccountDao(ReactiveEntityStore data) {
+ super(data);
+ }
+
+ public Single insert(AccountModel account) {
+ return data().insert(account);
+ }
+
+ public ReactiveResult get(UUID accountId) {
+ return data().select(AccountModel.class)
+ .where(AccountModel.ID.eq(accountId))
+ .get();
+ }
+
+ public ReactiveResult get(String address) {
+ return data().select(AccountModel.class)
+ .where(AccountModel.ADDRESS.eq(address))
+ .get();
+ }
+
+ public ReactiveResult getAll() {
+ return data().select(AccountModel.class)
+ .get();
+ }
+
+ public Single delete(UUID accountId) {
+ return data().delete(AccountModel.class)
+ .where(AccountModel.ID.eq(accountId))
+ .get().single();
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/dao/DirectChatDao.java b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/DirectChatDao.java
new file mode 100644
index 0000000..6f7f68b
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/DirectChatDao.java
@@ -0,0 +1,48 @@
+package org.mercury_im.messenger.data.repository.dao;
+
+import org.mercury_im.messenger.data.model.DirectChatModel;
+
+import java.util.UUID;
+
+import javax.inject.Inject;
+
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveResult;
+
+public class DirectChatDao extends RequeryDao {
+
+ @Inject
+ public DirectChatDao(ReactiveEntityStore data) {
+ super(data);
+ }
+
+ public Single insert(DirectChatModel chat) {
+ return data().insert(chat);
+ }
+
+ public ReactiveResult get(UUID chatId) {
+ return data().select(DirectChatModel.class)
+ .where(DirectChatModel.ID.eq(chatId))
+ .get();
+ }
+
+ public ReactiveResult getByPeer(UUID peerId) {
+ return data().select(DirectChatModel.class)
+ .where(DirectChatModel.PEER_ID.eq(peerId))
+ .get();
+ }
+
+ public ReactiveResult getAll() {
+ return data().select(DirectChatModel.class)
+ .get();
+ }
+
+ public Single delete(UUID chatId) {
+ return data().delete(DirectChatModel.class)
+ .where(DirectChatModel.ID.eq(chatId))
+ .get()
+ .single();
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/dao/GroupChatDao.java b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/GroupChatDao.java
new file mode 100644
index 0000000..9997ceb
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/GroupChatDao.java
@@ -0,0 +1,49 @@
+package org.mercury_im.messenger.data.repository.dao;
+
+import org.mercury_im.messenger.data.model.GroupChatModel;
+
+import java.util.UUID;
+
+import javax.inject.Inject;
+
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveResult;
+
+public class GroupChatDao extends RequeryDao {
+
+ @Inject
+ public GroupChatDao(ReactiveEntityStore data) {
+ super(data);
+ }
+
+ public Single insert(GroupChatModel chat) {
+ return data().insert(chat);
+ }
+
+ public ReactiveResult get(UUID chatId) {
+ return data().select(GroupChatModel.class)
+ .where(GroupChatModel.ID.eq(chatId))
+ .get();
+ }
+
+ public ReactiveResult get(UUID accountId, String address) {
+ return data().select(GroupChatModel.class)
+ .where(GroupChatModel.ACCOUNT_ID.eq(accountId))
+ .and(GroupChatModel.ADDRESS.eq(address))
+ .get();
+ }
+
+ public ReactiveResult getAll() {
+ return data().select(GroupChatModel.class)
+ .get();
+ }
+
+ public Single delete(UUID chatId) {
+ return data().delete(GroupChatModel.class)
+ .where(GroupChatModel.ID.eq(chatId))
+ .get()
+ .single();
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/dao/MessageDao.java b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/MessageDao.java
new file mode 100644
index 0000000..c9c6a5f
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/MessageDao.java
@@ -0,0 +1,52 @@
+package org.mercury_im.messenger.data.repository.dao;
+
+import org.mercury_im.messenger.data.model.DirectChatModel;
+import org.mercury_im.messenger.data.model.DirectMessagesRelation;
+import org.mercury_im.messenger.data.model.GroupChatModel;
+import org.mercury_im.messenger.data.model.GroupMessagesRelation;
+import org.mercury_im.messenger.data.model.MessageModel;
+
+import java.util.UUID;
+
+import io.reactivex.Single;
+import io.requery.Persistable;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveResult;
+
+public class MessageDao extends RequeryDao {
+
+ private final DirectChatDao directChatDao;
+ private final GroupChatDao groupChatDao;
+
+ public MessageDao(ReactiveEntityStore data) {
+ super(data);
+ this.directChatDao = new DirectChatDao(data);
+ this.groupChatDao = new GroupChatDao(data);
+ }
+
+ public Single insert(MessageModel message) {
+ return data().insert(message);
+ }
+
+ public ReactiveResult get(UUID messageId) {
+ return data().select(MessageModel.class)
+ .where(MessageModel.ID.eq(messageId))
+ .get();
+ }
+
+
+
+ private DirectMessagesRelation toRelation(DirectChatModel chat, MessageModel message) {
+ DirectMessagesRelation relation = new DirectMessagesRelation();
+ relation.setChat(chat);
+ relation.setMessage(message);
+ return relation;
+ }
+
+ private GroupMessagesRelation toRelation(GroupChatModel chat, MessageModel message) {
+ GroupMessagesRelation relation = new GroupMessagesRelation();
+ relation.setChat(chat);
+ relation.setMessage(message);
+ return relation;
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/dao/RequeryDao.java b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/RequeryDao.java
new file mode 100644
index 0000000..813ef3b
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/dao/RequeryDao.java
@@ -0,0 +1,17 @@
+package org.mercury_im.messenger.data.repository.dao;
+
+import io.requery.Persistable;
+import io.requery.reactivex.ReactiveEntityStore;
+
+public abstract class RequeryDao {
+
+ private final ReactiveEntityStore data;
+
+ public RequeryDao(ReactiveEntityStore data) {
+ this.data = data;
+ }
+
+ public ReactiveEntityStore data() {
+ return data;
+ }
+}
diff --git a/data/src/main/java/org/mercury_im/messenger/data/repository/package-info.java b/data/src/main/java/org/mercury_im/messenger/data/repository/package-info.java
new file mode 100644
index 0000000..3dfd22e
--- /dev/null
+++ b/data/src/main/java/org/mercury_im/messenger/data/repository/package-info.java
@@ -0,0 +1,11 @@
+/**
+ * The repository package contains implementations of the repositories defined in the domain
+ * module.
+ *
+ * While the data module uses requery to store data in an SQL database, the repositories use
+ * mappers defined in the mapping package to map the requery models to entities.
+ *
+ * Since the application itself only ever uses entities, it doesn't have to know about the database
+ * at all.
+ */
+package org.mercury_im.messenger.data.repository;
diff --git a/data/src/test/java/org/mercury_im/messenger/data/di/InMemoryDatabaseComponent.java b/data/src/test/java/org/mercury_im/messenger/data/di/InMemoryDatabaseComponent.java
new file mode 100644
index 0000000..37e71d6
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/di/InMemoryDatabaseComponent.java
@@ -0,0 +1,18 @@
+package org.mercury_im.messenger.data.di;
+
+import org.mercury_im.messenger.data.repository.AccountRepositoryTest;
+
+import javax.inject.Singleton;
+
+import dagger.Component;
+
+@Component(modules = {
+ RepositoryModule.class,
+ TestDatabaseModule.class,
+ TestingSchedulerModule.class
+})
+@Singleton
+public interface InMemoryDatabaseComponent {
+
+ void inject(AccountRepositoryTest test);
+}
diff --git a/data/src/test/java/org/mercury_im/messenger/data/di/MappingTestComponent.java b/data/src/test/java/org/mercury_im/messenger/data/di/MappingTestComponent.java
new file mode 100644
index 0000000..6c8ff56
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/di/MappingTestComponent.java
@@ -0,0 +1,21 @@
+package org.mercury_im.messenger.data.di;
+
+
+import org.mercury_im.messenger.data.mapping.AccountMappingTest;
+import org.mercury_im.messenger.data.mapping.EntityCapsMappingTest;
+import org.mercury_im.messenger.data.mapping.PeerMappingTest;
+
+import javax.inject.Singleton;
+
+import dagger.Component;
+
+@Component(modules = MappingModule.class)
+@Singleton
+public interface MappingTestComponent {
+
+ void inject(AccountMappingTest test);
+
+ void inject(PeerMappingTest test);
+
+ void inject(EntityCapsMappingTest test);
+}
diff --git a/data/src/test/java/org/mercury_im/messenger/data/di/TestDatabaseModule.java b/data/src/test/java/org/mercury_im/messenger/data/di/TestDatabaseModule.java
new file mode 100644
index 0000000..e16e621
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/di/TestDatabaseModule.java
@@ -0,0 +1,59 @@
+package org.mercury_im.messenger.data.di;
+
+import org.mercury_im.messenger.data.model.Models;
+import org.sqlite.SQLiteConfig;
+import org.sqlite.SQLiteDataSource;
+
+import java.io.PrintWriter;
+import java.sql.SQLException;
+
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+import io.requery.Persistable;
+import io.requery.cache.WeakEntityCache;
+import io.requery.meta.EntityModel;
+import io.requery.reactivex.ReactiveEntityStore;
+import io.requery.reactivex.ReactiveSupport;
+import io.requery.sql.Configuration;
+import io.requery.sql.ConfigurationBuilder;
+import io.requery.sql.EntityDataStore;
+import io.requery.sql.SchemaModifier;
+import io.requery.sql.TableCreationMode;
+
+@Module
+public class TestDatabaseModule {
+
+ @Provides
+ @Singleton
+ public static ReactiveEntityStore provideInMemoryEntityStore() {
+ EntityModel model = Models.DEFAULT;
+
+ SQLiteDataSource dataSource = new SQLiteDataSource();
+ try {
+ dataSource.setLogWriter(new PrintWriter(System.out));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ }
+ dataSource.setUrl("jdbc:sqlite:/tmp/mercury_testing.db");
+ dataSource.setDatabaseName("testing");
+ SQLiteConfig config = new SQLiteConfig();
+ config.setDateClass("TEXT");
+ dataSource.setConfig(config);
+ // Turn on foreign keys support.
+ // NOTE: Do it after setConfig, or setConfig will overwrite this setting
+ dataSource.setEnforceForeignKeys(true);
+
+ SchemaModifier modifier = new SchemaModifier(dataSource, model);
+ modifier.createTables(TableCreationMode.DROP_CREATE);
+
+ Configuration configuration = new ConfigurationBuilder(dataSource, model)
+ .setEntityCache(new WeakEntityCache())
+ .useDefaultLogging()
+ .build();
+
+ EntityDataStore dataStore = new EntityDataStore<>(configuration);
+ return ReactiveSupport.toReactiveStore(dataStore);
+ }
+}
diff --git a/data/src/test/java/org/mercury_im/messenger/data/di/TestingSchedulerModule.java b/data/src/test/java/org/mercury_im/messenger/data/di/TestingSchedulerModule.java
new file mode 100644
index 0000000..4216a43
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/di/TestingSchedulerModule.java
@@ -0,0 +1,29 @@
+package org.mercury_im.messenger.data.di;
+
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import javax.inject.Named;
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+import io.reactivex.Scheduler;
+import io.reactivex.schedulers.Schedulers;
+
+@Module
+public class TestingSchedulerModule {
+
+ @Provides
+ @Singleton
+ @Named(value = ThreadUtils.SCHEDULER_IO)
+ public static Scheduler provideSubscriberScheduler() {
+ return Schedulers.io();
+ }
+
+ @Provides
+ @Singleton
+ @Named(value = ThreadUtils.SCHEDULER_UI)
+ public static Scheduler provideObserverScheduler() {
+ return Schedulers.trampoline();
+ }
+}
diff --git a/data/src/test/java/org/mercury_im/messenger/data/mapping/AccountMappingTest.java b/data/src/test/java/org/mercury_im/messenger/data/mapping/AccountMappingTest.java
new file mode 100644
index 0000000..b2bdb99
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/mapping/AccountMappingTest.java
@@ -0,0 +1,67 @@
+package org.mercury_im.messenger.data.mapping;
+
+import org.junit.Test;
+import org.mercury_im.messenger.data.di.DaggerMappingTestComponent;
+import org.mercury_im.messenger.data.model.AccountModel;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.IAccount;
+
+import java.lang.reflect.Field;
+
+import javax.inject.Inject;
+
+import io.requery.proxy.EntityProxy;
+
+import static junit.framework.TestCase.assertEquals;
+import static junit.framework.TestCase.assertNotSame;
+
+public class AccountMappingTest {
+
+ @Inject
+ AccountMapping accountMapping;
+
+ public static final Account ACCOUNT_MISSION_CONTROL;
+ public static final Account ACCOUNT_LITTLE_JOE;
+
+ static {
+ ACCOUNT_MISSION_CONTROL = new IAccount();
+ ACCOUNT_MISSION_CONTROL.setAddress("mission-controll@planet.earth");
+ ACCOUNT_MISSION_CONTROL.setEnabled(true);
+ ACCOUNT_MISSION_CONTROL.setPassword("notBecauseTheyAreEasy");
+
+ ACCOUNT_LITTLE_JOE = new IAccount();
+ ACCOUNT_LITTLE_JOE.setAddress("little-joe@planet.earth");
+ ACCOUNT_LITTLE_JOE.setEnabled(false);
+ ACCOUNT_LITTLE_JOE.setPassword("butBecauseTheyAreHard");
+ }
+
+ public AccountMappingTest() {
+ DaggerMappingTestComponent.create().inject(this);
+ }
+
+ @Test
+ public void entityToModel() {
+
+ AccountModel model = accountMapping.toModel(ACCOUNT_MISSION_CONTROL);
+
+ assertEquals(ACCOUNT_MISSION_CONTROL.getId(), model.getId());
+ assertEquals(ACCOUNT_MISSION_CONTROL.getAddress(), model.getAddress());
+ assertEquals(ACCOUNT_MISSION_CONTROL.getPassword(), model.getPassword());
+ assertEquals(ACCOUNT_MISSION_CONTROL.isEnabled(), model.isEnabled());
+ }
+
+ @Test
+ public void modelToEntity() throws NoSuchFieldException, IllegalAccessException {
+ AccountModel model = new AccountModel();
+ model.getId();
+ model.setAddress("model@entity.store");
+ model.setEnabled(true);
+ model.setPassword("12345");
+
+ Account entity = accountMapping.toEntity(model);
+
+ assertEquals(model.getId(), entity.getId());
+ assertEquals(model.getAddress(), entity.getAddress());
+ assertEquals(model.getPassword(), entity.getPassword());
+ }
+}
diff --git a/data/src/test/java/org/mercury_im/messenger/data/mapping/EntityCapsMappingTest.java b/data/src/test/java/org/mercury_im/messenger/data/mapping/EntityCapsMappingTest.java
new file mode 100644
index 0000000..f684810
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/mapping/EntityCapsMappingTest.java
@@ -0,0 +1,74 @@
+package org.mercury_im.messenger.data.mapping;
+
+import org.junit.Test;
+import org.mercury_im.messenger.data.di.DaggerMappingTestComponent;
+import org.mercury_im.messenger.data.model.EntityCapsModel;
+import org.mercury_im.messenger.entity.caps.EntityCapsRecord;
+import org.mercury_im.messenger.entity.caps.IEntityCapsRecord;
+
+import javax.inject.Inject;
+
+import static junit.framework.TestCase.assertEquals;
+
+public class EntityCapsMappingTest {
+
+ @Inject
+ EntityCapsMapping mapping;
+
+ public EntityCapsMappingTest() {
+ DaggerMappingTestComponent.create().inject(this);
+ }
+
+ @Test
+ public void mapEntityToModelTest() {
+ EntityCapsRecord entity = new IEntityCapsRecord();
+ entity.setNodeVer("thisisahash");
+ entity.setXml("");
+
+ EntityCapsModel model = mapping.toModel(entity);
+
+ assertEquals(entity.getNodeVer(), model.getNodeVer());
+ assertEquals(entity.getXml(), model.getXml());
+ }
+
+ @Test
+ public void mapModelToEntityTest() {
+ EntityCapsModel model = new EntityCapsModel();
+ model.setNodeVer("q07IKJEyjvHSyhy//CH0CxmKi8w=");
+ model.setXml("" +
+ " " +
+ " " +
+ " " +
+ " " +
+ " " +
+ " " +
+ " " +
+ " " +
+ " urn:xmpp:dataforms:softwareinfo" +
+ " " +
+ " " +
+ " ipv4" +
+ " ipv6" +
+ " " +
+ " " +
+ " Mac" +
+ " " +
+ " " +
+ " 10.5.1" +
+ " " +
+ " " +
+ " Psi" +
+ " " +
+ " " +
+ " 0.11" +
+ " " +
+ " " +
+ " ");
+
+ EntityCapsRecord entity = mapping.toEntity(model);
+
+ assertEquals(model.getNodeVer(), entity.getNodeVer());
+ assertEquals(model.getXml(), entity.getXml());
+ }
+}
diff --git a/data/src/test/java/org/mercury_im/messenger/data/mapping/PeerMappingTest.java b/data/src/test/java/org/mercury_im/messenger/data/mapping/PeerMappingTest.java
new file mode 100644
index 0000000..84745ca
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/mapping/PeerMappingTest.java
@@ -0,0 +1,49 @@
+package org.mercury_im.messenger.data.mapping;
+
+import org.junit.Test;
+import org.mercury_im.messenger.data.di.DaggerMappingTestComponent;
+import org.mercury_im.messenger.data.model.AccountModel;
+import org.mercury_im.messenger.data.model.PeerModel;
+import org.mercury_im.messenger.entity.contact.IPeer;
+import org.mercury_im.messenger.entity.contact.Peer;
+import org.mercury_im.messenger.entity.contact.SubscriptionDirection;
+
+import javax.inject.Inject;
+
+import static junit.framework.TestCase.assertEquals;
+
+public class PeerMappingTest {
+
+ @Inject
+ PeerMapping peerMapping;
+
+ public static final Peer PEER_GORDO;
+
+ static {
+ PEER_GORDO = new IPeer();
+ PEER_GORDO.setAccount(AccountMappingTest.ACCOUNT_MISSION_CONTROL);
+ PEER_GORDO.setAddress("gordo@big.joe");
+ PEER_GORDO.setName("Gordo");
+ PEER_GORDO.setSubscriptionDirection(SubscriptionDirection.both);
+ }
+
+ public PeerMappingTest() {
+ DaggerMappingTestComponent.create().inject(this);
+ }
+
+ @Test
+ public void entityToModel() {
+ PeerModel model = peerMapping.toModel(PEER_GORDO);
+ assertEquals(PEER_GORDO.getAddress(), model.getAddress());
+ assertEquals(PEER_GORDO.getAccount().getAddress(), model.getAccount().getAddress());
+ assertEquals(PEER_GORDO.getName(), model.getName());
+ }
+
+ @Test
+ public void modelToEntity() {
+ PeerModel model = new PeerModel();
+ model.setName("Gordo");
+ model.setAddress("gordo@big.joe");
+ model.setAccount(new AccountModel());
+ }
+}
diff --git a/data/src/test/java/org/mercury_im/messenger/data/repository/AccountRepositoryTest.java b/data/src/test/java/org/mercury_im/messenger/data/repository/AccountRepositoryTest.java
new file mode 100644
index 0000000..0aba3d7
--- /dev/null
+++ b/data/src/test/java/org/mercury_im/messenger/data/repository/AccountRepositoryTest.java
@@ -0,0 +1,131 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.junit.Test;
+import org.mercury_im.messenger.data.di.DaggerInMemoryDatabaseComponent;
+import org.mercury_im.messenger.data.di.InMemoryDatabaseComponent;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.IAccount;
+
+import java.util.NoSuchElementException;
+import java.util.UUID;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.inject.Inject;
+
+import io.reactivex.disposables.CompositeDisposable;
+import io.requery.Persistable;
+import io.requery.reactivex.ReactiveEntityStore;
+
+public class AccountRepositoryTest {
+
+ private static final Logger LOGGER = Logger.getLogger(AccountRepositoryTest.class.getName());
+
+ @Inject
+ ReactiveEntityStore dataStore;
+
+ @Inject
+ XmppAccountRepository accountRepository;
+
+ @Inject
+ XmppDirectChatRepository directChatRepository;
+
+ @Inject
+ XmppPeerRepository contactRepository;
+
+ @Inject
+ XmppMessageRepository messageRepository;
+
+ @Inject
+ public AccountRepositoryTest() {
+ InMemoryDatabaseComponent testComponent = DaggerInMemoryDatabaseComponent.builder()
+ .build();
+ testComponent.inject(this);
+ }
+
+ @Test
+ public void test() throws InterruptedException {
+ CompositeDisposable d = new CompositeDisposable();
+ d.add(accountRepository.observeAccounts()
+ .distinct(Account::getId)
+ .subscribe(a -> System.out.println("Observe: " + a.getAddress())));
+
+ Thread.sleep(100);
+
+ Account a1 = new IAccount();
+ a1.setAddress("a1@example.com");
+ a1.setPassword("a1a1a1");
+ a1.setEnabled(true);
+
+ d.add(accountRepository.insertAccount(a1).subscribe());
+
+ Thread.sleep(100);
+
+ Account a2 = new IAccount();
+ a2.setAddress("a2@example.com");
+ a2.setPassword("a2a2a2");
+ a2.setEnabled(false);
+
+ d.add(accountRepository.insertAccount(a2).subscribe());
+
+ Thread.sleep(100);
+
+ Account a3 = new IAccount();
+ a3.setAddress("a3@example.com");
+ a3.setPassword("a3a3a3");
+ a3.setEnabled(false);
+
+ d.add(accountRepository.insertAccount(a3).subscribe());
+
+ Thread.sleep(100);
+
+ a1.setAddress("a11@example.org");
+
+ d.add(accountRepository.updateAccount(a1).subscribe());
+
+ Thread.sleep(100);
+ }
+
+ @Test
+ public void observeDeletionTest() throws InterruptedException {
+ CompositeDisposable d = new CompositeDisposable();
+
+ UUID uuid = UUID.randomUUID();
+ d.add(accountRepository.observeAccount(uuid)
+ .subscribe(optAccount -> {
+ LOGGER.log(Level.INFO, "Changed");
+ if (!optAccount.isPresent()) {
+ LOGGER.log(Level.INFO, "Changed: Account " + uuid.toString() + " is not present.");
+ } else {
+ LOGGER.log(Level.INFO, "Changed: Account " + uuid.toString() + ": " + optAccount.getItem().toString());
+ }
+ }));
+
+ Account account = new IAccount(uuid);
+ account.setEnabled(true);
+ account.setAddress("hello@world");
+ account.setPassword("wooooooh");
+
+ d.add(accountRepository.insertAccount(account)
+ .subscribe(insert -> LOGGER.log(Level.INFO, "Inserted."),
+ error -> LOGGER.log(Level.SEVERE, "Inserted", error)));
+
+ Thread.sleep(100);
+
+ d.add(accountRepository.deleteAccount(uuid)
+ .subscribe(() -> LOGGER.log(Level.INFO, "Deleted")));
+
+ Thread.sleep(1000);
+ d.dispose();
+ }
+
+ @Test(expected = NoSuchElementException.class)
+ public void updateMissingEntityFails() {
+ Account missingAccount = new IAccount();
+ missingAccount.setAddress("this@account.is.missing");
+ missingAccount.setPassword("inTheDatabase");
+
+ accountRepository.updateAccount(missingAccount)
+ .blockingGet();
+ }
+}
diff --git a/domain/.gitignore b/domain/.gitignore
new file mode 100644
index 0000000..60bf954
--- /dev/null
+++ b/domain/.gitignore
@@ -0,0 +1,2 @@
+/build
+testcredentials.properties
diff --git a/domain/build.gradle b/domain/build.gradle
new file mode 100644
index 0000000..187c55a
--- /dev/null
+++ b/domain/build.gradle
@@ -0,0 +1,32 @@
+apply plugin: 'java-library'
+
+dependencies {
+
+ implementation project(':entity')
+
+ // Smack
+ // Not all of those are needed, but it may be a good idea to define those versions explicitly
+ api "org.igniterealtime.smack:smack-core:$smackCoreVersion"
+ api "org.igniterealtime.smack:smack-experimental:$smackExperimentalVersion"
+ api "org.igniterealtime.smack:smack-extensions:$smackExtensionsVersion"
+ api "org.igniterealtime.smack:smack-im:$smackImVersion"
+ api "org.igniterealtime.smack:smack-tcp:$smackTcpVersion"
+
+ testImplementation "org.igniterealtime.smack:smack-java7:$smackJava7Version"
+
+ // RxJava2
+ implementation "io.reactivex.rxjava2:rxjava:$rxJava2Version"
+
+ // Dagger 2 for dependency injection
+ implementation "com.google.dagger:dagger:$daggerVersion"
+ annotationProcessor "com.google.dagger:dagger-compiler:$daggerVersion"
+
+ compileOnly 'org.projectlombok:lombok:1.18.10'
+ annotationProcessor 'org.projectlombok:lombok:1.18.10'
+
+ // JUnit for testing
+ testImplementation "junit:junit:$junitVersion"
+}
+
+sourceCompatibility = "8"
+targetCompatibility = "8"
diff --git a/domain/src/main/java/org/mercury_im/messenger/ClientStateListener.java b/domain/src/main/java/org/mercury_im/messenger/ClientStateListener.java
new file mode 100644
index 0000000..61964d0
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/ClientStateListener.java
@@ -0,0 +1,8 @@
+package org.mercury_im.messenger;
+
+public interface ClientStateListener {
+
+ void onClientInForeground();
+
+ void onClientInBackground();
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/MercurySchedulers.java b/domain/src/main/java/org/mercury_im/messenger/MercurySchedulers.java
new file mode 100644
index 0000000..a51d9c1
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/MercurySchedulers.java
@@ -0,0 +1,24 @@
+package org.mercury_im.messenger;
+
+import org.mercury_im.messenger.util.ThreadUtils;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+
+import io.reactivex.Scheduler;
+import lombok.Getter;
+
+public class MercurySchedulers {
+
+ @Getter
+ private final Scheduler subscriberScheduler;
+ @Getter
+ private final Scheduler observerScheduler;
+
+ @Inject
+ public MercurySchedulers(@Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
+ @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler) {
+ this.subscriberScheduler = subscriberScheduler;
+ this.observerScheduler = observerScheduler;
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/MessageCenter.java b/domain/src/main/java/org/mercury_im/messenger/MessageCenter.java
new file mode 100644
index 0000000..995f780
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/MessageCenter.java
@@ -0,0 +1,16 @@
+package org.mercury_im.messenger;
+
+import org.mercury_im.messenger.listener.IncomingDirectMessageListener;
+import org.mercury_im.messenger.entity.chat.Chat;
+import org.mercury_im.messenger.entity.message.Message;
+
+import io.reactivex.Completable;
+
+public interface MessageCenter {
+
+ Messenger getMessenger();
+
+ Completable sendMessage(Message message, C chat);
+
+ void addIncomingMessageListener(IncomingDirectMessageListener listener);
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/Messenger.java b/domain/src/main/java/org/mercury_im/messenger/Messenger.java
new file mode 100644
index 0000000..1e53829
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/Messenger.java
@@ -0,0 +1,86 @@
+package org.mercury_im.messenger;
+
+import org.jivesoftware.smack.SmackException;
+import org.jivesoftware.smackx.csi.ClientStateIndicationManager;
+import org.mercury_im.messenger.data.repository.Repositories;
+import org.mercury_im.messenger.usecase.AddAccount;
+import org.mercury_im.messenger.usecase.RosterStoreBinder;
+import org.mercury_im.messenger.xmpp.MercuryConnection;
+import org.mercury_im.messenger.xmpp.MercuryConnectionManager;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import io.reactivex.disposables.CompositeDisposable;
+import io.reactivex.schedulers.Schedulers;
+
+@Singleton
+public class Messenger implements ClientStateListener {
+
+ public static final String TAG = "MercuryIM";
+ private static final Logger LOGGER = Logger.getLogger(Messenger.class.getName());
+
+ private final MercuryConnectionManager connectionManager;
+ private final Repositories repositories;
+
+ private CompositeDisposable disposable = new CompositeDisposable();
+
+ @Inject
+ public Messenger(Repositories repositories, MercuryConnectionManager connectionManager) {
+ this.repositories = repositories;
+ this.connectionManager = connectionManager;
+ performInitialLogin();
+ }
+
+ public MercuryConnectionManager getConnectionManager() {
+ return connectionManager;
+ }
+
+ public void performInitialLogin() {
+ LOGGER.log(Level.INFO, "Perform initial login.");
+ disposable.add(repositories.getAccountRepository().observeAllAccounts().firstOrError()
+ .subscribeOn(Schedulers.newThread())
+ .subscribe(connectionManager::registerConnections));
+ }
+
+ public AddAccount addAccount() {
+ return new AddAccount(repositories.getAccountRepository(), this.connectionManager);
+ }
+
+ // CSI
+
+ @Override
+ public void onClientInForeground() {
+ LOGGER.log(Level.INFO, "CSI: active");
+ for (MercuryConnection connection : connectionManager.getConnections()) {
+ tryCsiActive(connection);
+ }
+ }
+
+ @Override
+ public void onClientInBackground() {
+ LOGGER.log(Level.INFO, "CSI: inactive");
+ for (MercuryConnection connection : connectionManager.getConnections()) {
+ tryCsiInactive(connection);
+ }
+ }
+
+ private void tryCsiActive(MercuryConnection connection) {
+ try {
+ ClientStateIndicationManager.active(connection.getConnection());
+ } catch (SmackException.NotConnectedException | InterruptedException e) {
+ LOGGER.log(Level.WARNING, "Sending CSI state 'active' failed.", e);
+ }
+ }
+
+ private void tryCsiInactive(MercuryConnection connection) {
+ try {
+ ClientStateIndicationManager.inactive(connection.getConnection());
+ } catch (SmackException.NotConnectedException | InterruptedException e) {
+ LOGGER.log(Level.WARNING, "Sending CSI state 'inactive' failed.", e);
+ }
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/account/error/PasswordError.java b/domain/src/main/java/org/mercury_im/messenger/account/error/PasswordError.java
new file mode 100644
index 0000000..6efc818
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/account/error/PasswordError.java
@@ -0,0 +1,8 @@
+package org.mercury_im.messenger.account.error;
+
+public enum PasswordError {
+ none,
+ emptyPassword,
+ invalidPassword,
+ incorrectPassword
+}
\ No newline at end of file
diff --git a/domain/src/main/java/org/mercury_im/messenger/account/error/UsernameError.java b/domain/src/main/java/org/mercury_im/messenger/account/error/UsernameError.java
new file mode 100644
index 0000000..33b9030
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/account/error/UsernameError.java
@@ -0,0 +1,8 @@
+package org.mercury_im.messenger.account.error;
+
+public enum UsernameError {
+ none,
+ emptyUsername,
+ invalidUsername,
+ unknownUsername
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/data/repository/AccountRepository.java b/domain/src/main/java/org/mercury_im/messenger/data/repository/AccountRepository.java
new file mode 100644
index 0000000..18eae4a
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/data/repository/AccountRepository.java
@@ -0,0 +1,48 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.util.Optional;
+
+import java.util.List;
+import java.util.UUID;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+
+public interface AccountRepository {
+
+ Single insertAccount(Account account);
+
+ default Observable> observeAccount(Account account) {
+ return observeAccount(account.getId());
+ }
+
+ Observable> observeAccount(UUID accountId);
+
+ default Maybe getAccount(Account account) {
+ return getAccount(account.getId());
+ }
+
+ Maybe getAccount(UUID accountId);
+
+ Observable> observeAccountByAddress(String address);
+
+ Maybe getAccountByAddress(String address);
+
+ Observable> observeAllAccounts();
+
+ Observable observeAccounts();
+
+ Single updateAccount(Account account);
+
+ Single upsertAccount(Account account);
+
+ default Completable deleteAccount(Account account) {
+ return deleteAccount(account.getId());
+ }
+
+ Completable deleteAccount(UUID accountId);
+
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/data/repository/DirectChatRepository.java b/domain/src/main/java/org/mercury_im/messenger/data/repository/DirectChatRepository.java
new file mode 100644
index 0000000..e15c442
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/data/repository/DirectChatRepository.java
@@ -0,0 +1,45 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.util.Optional;
+import org.mercury_im.messenger.entity.chat.DirectChat;
+import org.mercury_im.messenger.entity.contact.Peer;
+
+import java.util.List;
+import java.util.UUID;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+
+public interface DirectChatRepository {
+
+ Single insertDirectChat(DirectChat chat);
+
+ default Observable> observeDirectChat(DirectChat chat) {
+ return observeDirectChat(chat.getId());
+ }
+
+ Observable> observeDirectChat(UUID chatId);
+
+ Maybe getDirectChat(UUID chatId);
+
+ Single getOrCreateChatWithPeer(Peer peer);
+
+ Observable> observeDirectChatByPeer(Peer peer);
+
+ Maybe getDirectChatByPeer(Peer peer);
+
+ Observable> observeAllDirectChats();
+
+ Single updateDirectChat(DirectChat chat);
+
+ Single upsertDirectChat(DirectChat chat);
+
+ default Completable deleteDirectChat(DirectChat chat) {
+ return deleteDirectChat(chat.getId());
+ }
+
+ Completable deleteDirectChat(UUID chatId);
+
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/data/repository/EntityCapsRepository.java b/domain/src/main/java/org/mercury_im/messenger/data/repository/EntityCapsRepository.java
new file mode 100644
index 0000000..6273464
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/data/repository/EntityCapsRepository.java
@@ -0,0 +1,20 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.entity.caps.EntityCapsRecord;
+
+import java.util.Map;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+
+public interface EntityCapsRepository {
+
+ Observable> observeAllEntityCapsRecords();
+
+ Observable observeEntityCapsRecords();
+
+ Maybe maybeGetEntityCapsRecord(String nodeVer);
+
+ Completable insertEntityCapsRecord(EntityCapsRecord entityCapsRecord);
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/data/repository/GroupChatRepository.java b/domain/src/main/java/org/mercury_im/messenger/data/repository/GroupChatRepository.java
new file mode 100644
index 0000000..0c79ef2
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/data/repository/GroupChatRepository.java
@@ -0,0 +1,55 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.chat.GroupChat;
+import org.mercury_im.messenger.util.Optional;
+
+import java.util.List;
+import java.util.UUID;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+
+public interface GroupChatRepository {
+
+ Single insertGroupChat(GroupChat chat);
+
+ Observable> observeGroupChat(UUID chatId);
+
+ Maybe getGroupChat(UUID chatId);
+
+ Single getOrCreateGroupChat(Account account, String roomAddress);
+
+ default Observable> observeGroupChatByRoomAddress(Account account, String roomAddress) {
+ return observeGroupChatByRoomAddress(account.getId(), roomAddress);
+ }
+
+ Observable> observeGroupChatByRoomAddress(UUID accountId, String roomAddress);
+
+ default Maybe getGroupChatByRoomAddress(Account account, String roomAddress) {
+ return getGroupChatByRoomAddress(account.getId(), roomAddress);
+ }
+
+ Maybe getGroupChatByRoomAddress(UUID accountId, String roomAddress);
+
+ Observable> observeAllGroupChats();
+
+ default Observable> observeAllGroupChatsOfAccount(Account account) {
+ return observeAllGroupChatsOfAccount(account.getId());
+ }
+
+ Observable> observeAllGroupChatsOfAccount(UUID accountId);
+
+ Single updateGroupChat(GroupChat chat);
+
+ Single upsertGroupChat(GroupChat chat);
+
+ default Completable deleteGroupChat(GroupChat chat) {
+ return deleteGroupChat(chat.getId());
+ }
+
+ Completable deleteGroupChat(UUID chatId);
+
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/data/repository/MessageRepository.java b/domain/src/main/java/org/mercury_im/messenger/data/repository/MessageRepository.java
new file mode 100644
index 0000000..bed2cea
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/data/repository/MessageRepository.java
@@ -0,0 +1,36 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.entity.chat.DirectChat;
+import org.mercury_im.messenger.entity.chat.GroupChat;
+import org.mercury_im.messenger.entity.message.Message;
+
+import java.util.List;
+
+import io.reactivex.Completable;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+
+public interface MessageRepository {
+
+ Single insertMessage(DirectChat chat, Message message);
+
+ Single insertMessage(GroupChat chat, Message message);
+
+ Observable> observeMessages(DirectChat chat);
+
+ Observable> observeMessages(GroupChat chat);
+
+ Observable> findMessagesWithBody(String body);
+
+ Observable> findMessagesWithBody(DirectChat chat, String body);
+
+ Observable> findMessagesWithBody(GroupChat chat, String body);
+
+ Single upsertMessage(DirectChat chat, Message message);
+
+ Single upsertMessage(GroupChat chat, Message message);
+
+ Single updateMessage(Message message);
+
+ Completable deleteMessage(Message message);
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/data/repository/PeerRepository.java b/domain/src/main/java/org/mercury_im/messenger/data/repository/PeerRepository.java
new file mode 100644
index 0000000..d3585c7
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/data/repository/PeerRepository.java
@@ -0,0 +1,58 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.util.Optional;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.contact.Peer;
+
+import java.util.List;
+import java.util.UUID;
+
+import io.reactivex.Completable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+
+public interface PeerRepository {
+
+ Single insertPeer(Peer Peer);
+
+ default Observable> observePeer(Peer peer) {
+ return observePeer(peer.getId());
+ }
+
+ Observable> observePeer(UUID PeerId);
+
+ Maybe getPeer(UUID PeerId);
+
+ default Observable> observePeerByAddress(Account account, String address) {
+ return observePeerByAddress(account.getId(), address);
+ }
+
+ Observable> observePeerByAddress(UUID accountId, String address);
+
+ default Maybe getPeerByAddress(Account account, String address) {
+ return getPeerByAddress(account.getId(), address);
+ }
+
+ Maybe getPeerByAddress(UUID accountId, String address);
+
+ Single getOrCreatePeer(Account account, String address);
+
+ Single getOrCreatePeer(UUID accountId, String address);
+
+ Observable> observeAllPeers();
+
+ default Observable> observeAllContactsOfAccount(Account account) {
+ return observeAllContactsOfAccount(account.getId());
+ }
+
+ Observable> observeAllContactsOfAccount(UUID accountId);
+
+ Single updatePeer(Peer Peer);
+
+ Single upsertPeer(Peer Peer);
+
+ Completable deletePeer(Peer Peer);
+
+ Completable deletePeer(UUID accountId, String address);
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/data/repository/Repositories.java b/domain/src/main/java/org/mercury_im/messenger/data/repository/Repositories.java
new file mode 100644
index 0000000..bacdfd1
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/data/repository/Repositories.java
@@ -0,0 +1,56 @@
+package org.mercury_im.messenger.data.repository;
+
+import org.mercury_im.messenger.entity.caps.EntityCapsRecord;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+@Singleton
+public class Repositories {
+
+ private final AccountRepository accountRepository;
+ private final DirectChatRepository directChatRepository;
+ private final GroupChatRepository groupChatRepository;
+ private final MessageRepository messageRepository;
+ private final PeerRepository peerRepository;
+ private final EntityCapsRepository entityCapsRepository;
+
+ @Inject
+ public Repositories(AccountRepository accountRepository,
+ DirectChatRepository directChatRepository,
+ GroupChatRepository groupChatRepository,
+ MessageRepository messageRepository,
+ PeerRepository peerRepository,
+ EntityCapsRepository entityCapsRepository) {
+ this.accountRepository = accountRepository;
+ this.directChatRepository = directChatRepository;
+ this.groupChatRepository = groupChatRepository;
+ this.messageRepository = messageRepository;
+ this.peerRepository = peerRepository;
+ this.entityCapsRepository = entityCapsRepository;
+ }
+
+ public AccountRepository getAccountRepository() {
+ return accountRepository;
+ }
+
+ public DirectChatRepository getDirectChatRepository() {
+ return directChatRepository;
+ }
+
+ public GroupChatRepository getGroupChatRepository() {
+ return groupChatRepository;
+ }
+
+ public MessageRepository getMessageRepository() {
+ return messageRepository;
+ }
+
+ public PeerRepository getPeerRepository() {
+ return peerRepository;
+ }
+
+ public EntityCapsRepository getEntityCapsRepository() {
+ return entityCapsRepository;
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/exception/IllegalUsernameException.java b/domain/src/main/java/org/mercury_im/messenger/exception/IllegalUsernameException.java
new file mode 100644
index 0000000..6e084dc
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/exception/IllegalUsernameException.java
@@ -0,0 +1,8 @@
+package org.mercury_im.messenger.exception;
+
+public class IllegalUsernameException extends RuntimeException {
+
+ public IllegalUsernameException(Throwable cause) {
+ super(cause);
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/listener/IncomingDirectMessageListener.java b/domain/src/main/java/org/mercury_im/messenger/listener/IncomingDirectMessageListener.java
new file mode 100644
index 0000000..3facc87
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/listener/IncomingDirectMessageListener.java
@@ -0,0 +1,11 @@
+package org.mercury_im.messenger.listener;
+
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.chat.DirectChat;
+import org.mercury_im.messenger.entity.message.Message;
+
+public interface IncomingDirectMessageListener {
+
+ void onIncomingDirectMessage(Account account, DirectChat chat, Message message);
+
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/listener/IncomingGroupChatMessageListener.java b/domain/src/main/java/org/mercury_im/messenger/listener/IncomingGroupChatMessageListener.java
new file mode 100644
index 0000000..3fc9af3
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/listener/IncomingGroupChatMessageListener.java
@@ -0,0 +1,11 @@
+package org.mercury_im.messenger.listener;
+
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.chat.GroupChat;
+import org.mercury_im.messenger.entity.message.Message;
+
+public interface IncomingGroupChatMessageListener {
+
+ void onIncomingDirectMessage(Account account, GroupChat chat, Message message);
+
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/listener/TypingEventListener.java b/domain/src/main/java/org/mercury_im/messenger/listener/TypingEventListener.java
new file mode 100644
index 0000000..3d8e037
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/listener/TypingEventListener.java
@@ -0,0 +1,9 @@
+package org.mercury_im.messenger.listener;
+
+import org.mercury_im.messenger.entity.chat.Chat;
+import org.mercury_im.messenger.entity.event.TypingEvent;
+
+public interface TypingEventListener {
+
+ void onTypingEventReceived(Chat chat, TypingEvent typingEvent);
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/logging/Tags.java b/domain/src/main/java/org/mercury_im/messenger/logging/Tags.java
new file mode 100644
index 0000000..6ace309
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/logging/Tags.java
@@ -0,0 +1,9 @@
+package org.mercury_im.messenger.logging;
+
+public class Tags {
+
+ public static final String TAG_XMPP = "MercuryXMPP";
+ public static final String TAG_DB = "MercuryDB";
+ public static final String TAG_ANDROID = "MercuryAndroid";
+ public static final String TAG_DOMAIN = "MercuryDomain";
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/store/MercuryEntityCapsStore.java b/domain/src/main/java/org/mercury_im/messenger/store/MercuryEntityCapsStore.java
new file mode 100644
index 0000000..10700a1
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/store/MercuryEntityCapsStore.java
@@ -0,0 +1,63 @@
+package org.mercury_im.messenger.store;
+
+import org.jivesoftware.smack.util.PacketParserUtils;
+import org.jivesoftware.smack.xml.XmlPullParser;
+import org.jivesoftware.smackx.caps.cache.EntityCapsPersistentCache;
+import org.jivesoftware.smackx.disco.packet.DiscoverInfo;
+import org.mercury_im.messenger.data.repository.EntityCapsRepository;
+import org.mercury_im.messenger.entity.caps.EntityCapsRecord;
+import org.mercury_im.messenger.entity.caps.IEntityCapsRecord;
+import org.mercury_im.messenger.logging.Tags;
+
+import java.io.StringReader;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import io.reactivex.disposables.CompositeDisposable;
+
+@Singleton
+public class MercuryEntityCapsStore implements EntityCapsPersistentCache {
+
+ private static final Logger LOGGER = Logger.getLogger(Tags.TAG_DOMAIN);
+
+ private final CompositeDisposable disposable = new CompositeDisposable();
+ private final EntityCapsRepository repository;
+
+ @Inject
+ public MercuryEntityCapsStore(EntityCapsRepository entityCapsRepository) {
+ this.repository = entityCapsRepository;
+ }
+
+ @Override
+ public void addDiscoverInfoByNodePersistent(String nodeVer, DiscoverInfo info) {
+ LOGGER.log(Level.INFO, "MercuryEntityCapsStore: addDiscoverInfoByNodePersistent: " + nodeVer);
+ EntityCapsRecord record = new IEntityCapsRecord();
+ record.setNodeVer(nodeVer);
+ record.setXml(info.toXML().toString());
+
+ disposable.add(repository.insertEntityCapsRecord(record).subscribe());
+ }
+
+ @Override
+ public DiscoverInfo lookup(String nodeVer) {
+ LOGGER.log(Level.INFO, "MercuryEntityCapsStore: lookup: " + nodeVer);
+ return repository.maybeGetEntityCapsRecord(nodeVer)
+ .map(this::parseDiscoverInfo)
+ .onErrorComplete()
+ .blockingGet();
+ }
+
+ private DiscoverInfo parseDiscoverInfo(EntityCapsRecord record) throws Exception {
+ XmlPullParser parser = PacketParserUtils.getParserFor(new StringReader(record.getXml()));
+ return (DiscoverInfo) PacketParserUtils.parseIQ(parser);
+ }
+
+ @Override
+ public void emptyCache() {
+ LOGGER.log(Level.INFO, "MercuryEntityCapsStore: emptyCache.");
+ // Not needed?
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/store/MercuryRosterStore.java b/domain/src/main/java/org/mercury_im/messenger/store/MercuryRosterStore.java
new file mode 100644
index 0000000..63e9450
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/store/MercuryRosterStore.java
@@ -0,0 +1,208 @@
+package org.mercury_im.messenger.store;
+
+import org.jivesoftware.smack.roster.packet.RosterPacket;
+import org.jivesoftware.smack.roster.rosterstore.RosterStore;
+import org.jxmpp.jid.Jid;
+import org.jxmpp.jid.impl.JidCreate;
+import org.mercury_im.messenger.data.repository.AccountRepository;
+import org.mercury_im.messenger.data.repository.PeerRepository;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.entity.contact.Peer;
+import org.mercury_im.messenger.entity.contact.SubscriptionDirection;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import io.reactivex.disposables.CompositeDisposable;
+import io.reactivex.schedulers.Schedulers;
+
+public class MercuryRosterStore implements RosterStore {
+
+ private static final Logger LOGGER = Logger.getLogger(RosterStore.class.getName());
+
+ private final PeerRepository peerRepository;
+ private final AccountRepository accountRepository;
+ private Account account;
+ private final CompositeDisposable disposable = new CompositeDisposable();
+
+ private final Map itemMap = new HashMap<>();
+ private String rosterVersion;
+
+ public MercuryRosterStore(Account account, PeerRepository rosterRepository, AccountRepository accountRepository) {
+ this.account = account;
+ this.peerRepository = rosterRepository;
+ this.accountRepository = accountRepository;
+ LOGGER.log(Level.INFO, "Construct Roster Store for " + account.getId());
+ }
+
+ public void subscribe() {
+ disposable.add(peerRepository.observeAllContactsOfAccount(account)
+ .observeOn(Schedulers.computation())
+ .subscribe(contactsList -> {
+ itemMap.clear();
+ for (Peer contactModel : contactsList) {
+ itemMap.put(contactModel.getAddress(), fromEntity(contactModel));
+ LOGGER.log(Level.INFO, "Populate itemMap with " + contactsList.size() + " items");
+
+ }
+ },
+ error -> LOGGER.log(Level.WARNING, "An error occurred while updating roster cache", error)));
+
+ /*
+ disposable.add(peerRepository.getRosterVersion(account)
+ .observeOn(Schedulers.computation())
+ .subscribe(
+ result -> setRosterVersion(result),
+ error -> LOGGER.log(Level.WARNING, "An error occurred updating cached roster version", error)));
+
+ */
+ }
+
+ public void unsubscribe() {
+ disposable.dispose();
+ }
+
+ private void setRosterVersion(String rosterVersion) {
+ this.rosterVersion = rosterVersion;
+ }
+
+ @Override
+ public List getEntries() {
+ return new ArrayList<>(itemMap.values());
+ }
+
+ @Override
+ public RosterPacket.Item getEntry(Jid bareJid) {
+ return itemMap.get(bareJid.asUnescapedString());
+ }
+
+ @Override
+ public String getRosterVersion() {
+ return rosterVersion != null ? rosterVersion : "";
+ }
+
+ @Override
+ public boolean addEntry(RosterPacket.Item item, String version) {
+ writeEntryToDatabase(item);
+ writeRosterVersionToDatabase(version);
+
+ return true;
+ }
+
+ private void writeEntryToDatabase(RosterPacket.Item item) {
+ disposable.add(peerRepository.getOrCreatePeer(account, item.getJid().asUnescapedString())
+ .map(peer -> toEntity(item, peer))
+ .flatMap(peerRepository::upsertPeer)
+ .subscribe(
+ success -> LOGGER.log(Level.FINE, "Upserted contact model " + success + " successfully"),
+ error -> LOGGER.log(Level.WARNING, "An error occurred upserting contact " + item.getJid().asUnescapedString(), error)
+ ));
+ }
+
+ private void writeRosterVersionToDatabase(String version) {
+ /*
+ disposable.add(peerRepository.updateRosterVersion(account, version)
+ .subscribe(
+ success -> LOGGER.log(Level.FINE, "Upserted roster version to " + rosterVersion + " successfully"),
+ error -> LOGGER.log(Level.WARNING, "An error occurred upserting roster version", error)
+ ));
+ */
+ }
+
+ @Override
+ public boolean resetEntries(Collection items, String version) {
+ LOGGER.log(Level.INFO, "Reset Entries: " + Arrays.toString(items.toArray()));
+ // Update database
+ // TODO: Delete other contacts
+ for (RosterPacket.Item item : items) {
+ writeEntryToDatabase(item);
+ }
+
+ /*
+ disposable.add(peerRepository.updateRosterVersion(account, version)
+ .subscribe(
+ success -> LOGGER.log(Level.FINE, "Upserted roster version to " + rosterVersion + " successfully"),
+ error -> LOGGER.log(Level.WARNING, "An error occurred upserting roster version", error)
+ ));
+ */
+
+ return true;
+ }
+
+ @Override
+ public boolean removeEntry(Jid bareJid, String version) {
+ LOGGER.log(Level.INFO, "Remove entry " + bareJid.toString());
+
+ disposable.add(peerRepository.deletePeer(account.getId(), bareJid.asEntityBareJidOrThrow().asEntityBareJidString())
+ .subscribe(
+ () -> LOGGER.log(Level.FINE, "Deletion of contact " + bareJid.toString() + " successful"),
+ error -> LOGGER.log(Level.WARNING, "An error occurred deleting contact " + bareJid.toString(), error)
+ ));
+ /*
+ disposable.add(peerRepository.updateRosterVersion(account, version)
+ .subscribe(
+ success -> LOGGER.log(Level.FINE, "Upserted roster version to " + rosterVersion + " successfully"),
+ error -> LOGGER.log(Level.WARNING, "An error occurred upserting roster version", error)
+ ));
+ */
+ return true;
+ }
+
+ @Override
+ public void resetStore() {
+ LOGGER.log(Level.INFO, "Reset Store");
+
+ /*
+ disposable.add(peerRepository.deleteAllContactsOfAccount(account)
+ .subscribe(
+ success -> LOGGER.log(Level.FINE, "Successfully reset store."),
+ error -> LOGGER.log(Level.WARNING, "An error occurred resetting store", error)
+ ));
+ disposable.add(peerRepository.updateRosterVersion(account, "")
+ .subscribe(
+ success -> LOGGER.log(Level.FINE, "Successfully reset roster version"),
+ error -> LOGGER.log(Level.WARNING, "An error occurred resetting roster version", error)
+ ));
+ */
+ }
+
+ public RosterPacket.Item fromEntity(Peer contactModel) {
+ RosterPacket.Item item = new RosterPacket.Item(
+ JidCreate.entityBareFromOrThrowUnchecked(contactModel.getAddress()),
+ contactModel.getName());
+ if (contactModel.getSubscriptionDirection() != null) {
+ item.setItemType(convert(contactModel.getSubscriptionDirection()));
+ }
+ item.setApproved(contactModel.isSubscriptionApproved());
+ item.setSubscriptionPending(contactModel.isSubscriptionPending());
+ return item;
+ }
+
+ public Peer toEntity(RosterPacket.Item item, Peer peer) {
+ peer.setAccount(account);
+ peer.setAddress(item.getJid().asEntityBareJidOrThrow().asEntityBareJidString());
+ peer.setName(item.getName());
+ if (item.getItemType() != null) {
+ peer.setSubscriptionDirection(convert(item.getItemType()));
+ }
+ peer.setSubscriptionApproved(item.isApproved());
+ peer.setSubscriptionPending(item.isSubscriptionPending());
+
+ return peer;
+ }
+
+ public SubscriptionDirection convert(RosterPacket.ItemType type) {
+ return SubscriptionDirection.valueOf(type.toString());
+
+ }
+
+ public RosterPacket.ItemType convert(SubscriptionDirection direction) {
+ return RosterPacket.ItemType.fromString(direction.toString());
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/usecase/AddAccount.java b/domain/src/main/java/org/mercury_im/messenger/usecase/AddAccount.java
new file mode 100644
index 0000000..87d0b46
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/usecase/AddAccount.java
@@ -0,0 +1,82 @@
+package org.mercury_im.messenger.usecase;
+
+import org.mercury_im.messenger.data.repository.AccountRepository;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.xmpp.MercuryConnection;
+import org.mercury_im.messenger.xmpp.MercuryConnectionManager;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import io.reactivex.Completable;
+import io.reactivex.Single;
+import io.reactivex.disposables.CompositeDisposable;
+import io.reactivex.schedulers.Schedulers;
+
+public class AddAccount {
+
+ private static final Logger LOGGER = Logger.getLogger(AddAccount.class.getName());
+
+ private final AccountRepository accountRepository;
+ private final MercuryConnectionManager connectionManager;
+
+ private final CompositeDisposable disposable = new CompositeDisposable();
+
+ public AddAccount(AccountRepository accountRepository, MercuryConnectionManager connectionManager) {
+ this.accountRepository = accountRepository;
+ this.connectionManager = connectionManager;
+ }
+
+ public Completable execute(Account account) {
+ return loginAndStoreAccountIfSuccessful(account);
+ }
+
+ private Completable loginAndStoreAccountIfSuccessful(Account account) {
+ LOGGER.log(Level.INFO, "loginAndStoreIfSuccessful");
+ return logIntoAccount(account).flatMap(connection ->
+ insertEnabledAccountIntoDatabase(account).flatMap(insertedAccount ->
+ updateAccountIdInConnectionSingle(insertedAccount, connection)))
+ .map(this::addConnectionToMessenger)
+ .ignoreElement();
+ }
+
+ private Single logIntoAccount(Account account) {
+ return getOrCreateConnectionSingle(account)
+ .doAfterSuccess(con -> LogIntoAccount.with(con).executeAndPossiblyThrow())
+ .subscribeOn(Schedulers.io());
+ }
+
+ private Single getOrCreateConnectionSingle(Account account) {
+ return Single.fromCallable(() -> getOrCreateConnection(account));
+ }
+
+ private MercuryConnection getOrCreateConnection(Account account) {
+ MercuryConnection connection = connectionManager.getConnection(account);
+ if (connection == null) {
+ connection = new MercuryConnection(accountRepository, account);
+ }
+ return connection;
+ }
+
+ private Single insertEnabledAccountIntoDatabase(Account account) {
+ account.setEnabled(true);
+ return accountRepository.upsertAccount(account);
+ }
+
+ private Completable addConnectionToMessenger(MercuryConnection connection) {
+ return Completable.fromAction(() -> connectionManager.registerConnection(connection));
+ }
+
+ private Single updateAccountIdInConnectionSingle(Account account, MercuryConnection connection) {
+ return Single.fromCallable(() -> {
+ updateAccountIdInConnection(account, connection);
+ return connection;
+ });
+ }
+
+ private void updateAccountIdInConnection(Account account, MercuryConnection connection) {
+ if (connection != null) {
+ connection.getAccount().setId(account.getId());
+ }
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/usecase/AddAccountUseCase.java b/domain/src/main/java/org/mercury_im/messenger/usecase/AddAccountUseCase.java
new file mode 100644
index 0000000..6316d3d
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/usecase/AddAccountUseCase.java
@@ -0,0 +1,17 @@
+package org.mercury_im.messenger.usecase;
+
+import io.reactivex.Completable;
+
+public interface AddAccountUseCase {
+
+ AddAccountTask create();
+
+ Completable execute(AddAccountTask task);
+
+ interface AddAccountTask {
+
+ void setAddress(String address);
+
+ void setPassword(String password);
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/usecase/LogIntoAccount.java b/domain/src/main/java/org/mercury_im/messenger/usecase/LogIntoAccount.java
new file mode 100644
index 0000000..a3b74da
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/usecase/LogIntoAccount.java
@@ -0,0 +1,74 @@
+package org.mercury_im.messenger.usecase;
+
+import org.jivesoftware.smack.AbstractXMPPConnection;
+import org.jivesoftware.smack.SmackException;
+import org.jivesoftware.smack.XMPPException;
+import org.jivesoftware.smack.sasl.SASLErrorException;
+import org.mercury_im.messenger.xmpp.MercuryConnection;
+
+import java.io.IOException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import io.reactivex.Completable;
+import io.reactivex.Single;
+
+public class LogIntoAccount {
+
+ public enum ConnectionResult {
+ success,
+ credential_error,
+ server_error,
+ other_error
+ }
+
+ private static final Logger LOGGER = Logger.getLogger(LogIntoAccount.class.getName());
+
+ private final MercuryConnection connection;
+
+ private LogIntoAccount(MercuryConnection connection) {
+ this.connection = connection;
+ }
+
+ public static LogIntoAccount with(MercuryConnection connection) {
+ if (connection == null) {
+ throw new NullPointerException("MercuryConnection cannot be null.");
+ }
+ return new LogIntoAccount(connection);
+ }
+
+ public Completable executeAndPossiblyThrow() {
+ return Completable.fromAction(this::doAuthenticateIfNecessary);
+ }
+
+ public Single execute() {
+ return Single.fromCallable(this::authenticateIfNecessary);
+ }
+
+ private ConnectionResult authenticateIfNecessary() {
+ try {
+ doAuthenticateIfNecessary();
+ return ConnectionResult.success;
+ } catch (SASLErrorException e) {
+ LOGGER.log(Level.WARNING, "SASL Error while connecting to account " + connection.getAccount().getAddress(), e);
+ return ConnectionResult.credential_error;
+ } catch (SmackException.ConnectionException e) {
+ LOGGER.log(Level.WARNING, "Connectivity error while connecting to account " + connection.getAccount().getAddress(), e);
+ return ConnectionResult.server_error;
+ }
+ catch (IOException | XMPPException | SmackException | InterruptedException e) {
+ LOGGER.log(Level.WARNING, "Error connecting to account " + connection.getAccount().getAddress(), e);
+ return ConnectionResult.other_error;
+ }
+ }
+
+ private void doAuthenticateIfNecessary()
+ throws InterruptedException, XMPPException, SmackException, IOException {
+ if (!connection.getConnection().isAuthenticated()) {
+ LOGGER.log(Level.INFO, "Logging in");
+ ((AbstractXMPPConnection) connection.getConnection()).connect().login();
+ LOGGER.log(Level.INFO, "Login complete");
+ }
+ }
+
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/usecase/RosterStoreBinder.java b/domain/src/main/java/org/mercury_im/messenger/usecase/RosterStoreBinder.java
new file mode 100644
index 0000000..795916f
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/usecase/RosterStoreBinder.java
@@ -0,0 +1,33 @@
+package org.mercury_im.messenger.usecase;
+
+import org.jivesoftware.smack.roster.Roster;
+import org.mercury_im.messenger.data.repository.AccountRepository;
+import org.mercury_im.messenger.data.repository.PeerRepository;
+import org.mercury_im.messenger.entity.Account;
+import org.mercury_im.messenger.store.MercuryRosterStore;
+import org.mercury_im.messenger.xmpp.MercuryConnection;
+
+import javax.inject.Inject;
+
+public class RosterStoreBinder {
+
+ private final AccountRepository accountRepository;
+ private final PeerRepository peerRepository;
+
+ @Inject
+ public RosterStoreBinder(AccountRepository accountRepository, PeerRepository peerRepository) {
+ this.accountRepository = accountRepository;
+ this.peerRepository = peerRepository;
+ }
+
+ public void setRosterStoreOn(MercuryConnection connection) {
+ MercuryRosterStore store =
+ createRosterStore(connection.getAccount(), accountRepository, peerRepository);
+ Roster roster = Roster.getInstanceFor(connection.getConnection());
+ roster.setRosterStore(store);
+ }
+
+ private MercuryRosterStore createRosterStore(Account account, AccountRepository accountRepository, PeerRepository peerRepository) {
+ return new MercuryRosterStore(account, peerRepository, accountRepository);
+ }
+}
diff --git a/domain/src/main/java/org/mercury_im/messenger/util/DiffUtil.java b/domain/src/main/java/org/mercury_im/messenger/util/DiffUtil.java
new file mode 100644
index 0000000..ee472d7
--- /dev/null
+++ b/domain/src/main/java/org/mercury_im/messenger/util/DiffUtil.java
@@ -0,0 +1,57 @@
+package org.mercury_im.messenger.util;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class DiffUtil {
+
+ public static Diff diff(List before, List after) {
+ List removed = getRemovedItems(before, after);
+ List added = getAddedItems(before, after);
+ List retained = getRetainedItems(before, after);
+ return new Diff<>(added, retained, removed);
+ }
+
+ private static List getRemovedItems(List before, List after) {
+ List withoutRetained = new ArrayList<>(before);
+ withoutRetained.removeAll(after);
+ return withoutRetained;
+ }
+
+ private static