Синхронизация сообщений, чтения в группах, синхронизация ключей групп
This commit is contained in:
@@ -200,7 +200,7 @@ public class Boot {
|
|||||||
this.packetManager.registerExecutor(17, new Executor17GroupCreate());
|
this.packetManager.registerExecutor(17, new Executor17GroupCreate());
|
||||||
this.packetManager.registerExecutor(18, new Executor18GroupInfo());
|
this.packetManager.registerExecutor(18, new Executor18GroupInfo());
|
||||||
this.packetManager.registerExecutor(19, new Executor19GroupInviteInfo());
|
this.packetManager.registerExecutor(19, new Executor19GroupInviteInfo());
|
||||||
this.packetManager.registerExecutor(20, new Executor20GroupJoin());
|
this.packetManager.registerExecutor(20, new Executor20GroupJoin(this.packetManager));
|
||||||
this.packetManager.registerExecutor(21, new Executor21GroupLeave());
|
this.packetManager.registerExecutor(21, new Executor21GroupLeave());
|
||||||
this.packetManager.registerExecutor(22, new Executor22GroupBan());
|
this.packetManager.registerExecutor(22, new Executor22GroupBan());
|
||||||
this.packetManager.registerExecutor(24, new Executor24DeviceResolve(this.clientManager, this.eventManager, this.packetManager));
|
this.packetManager.registerExecutor(24, new Executor24DeviceResolve(this.clientManager, this.eventManager, this.packetManager));
|
||||||
|
|||||||
@@ -53,7 +53,7 @@ public class ClientManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Отправить пакет всем АВТОРИЗОВАННЫМ клиентам с публичным ключом publicKey
|
* Отправить пакет ВСЕМ АВТОРИЗОВАННЫМ клиентам с публичным ключом publicKey
|
||||||
* @param publicKey публичный ключ получателя
|
* @param publicKey публичный ключ получателя
|
||||||
* @param packet пакет для отправки
|
* @param packet пакет для отправки
|
||||||
* @throws ProtocolException если произошла ошибка при отправке пакета клиенту
|
* @throws ProtocolException если произошла ошибка при отправке пакета клиенту
|
||||||
@@ -83,6 +83,34 @@ public class ClientManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Отправить пакет всем клиентам с публичными ключом как у client, кроме клиента client, который является отправителем и не должен получать этот пакет
|
||||||
|
* @param client клиент
|
||||||
|
* @param packet пакет для отправки
|
||||||
|
* @throws ProtocolException
|
||||||
|
*/
|
||||||
|
public void retranslate(Client client, Packet packet) throws ProtocolException{
|
||||||
|
ECIAuthentificate eciAuthentificate = client.getTag(ECIAuthentificate.class);
|
||||||
|
HashSet<Client> clients = this.clientIndexer
|
||||||
|
.getClients(ECIAuthentificate.class, "publicKey", eciAuthentificate.getPublicKey());
|
||||||
|
if(clients == null){
|
||||||
|
/**
|
||||||
|
* Нет авторизованных сессий с таким публичным ключом
|
||||||
|
*/
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
for(Client c : clients){
|
||||||
|
/**
|
||||||
|
* Проходим по всем устройствам с таким публичным ключом и ретранслируем им пакет, кроме того устройства что
|
||||||
|
* отправило пакет
|
||||||
|
*/
|
||||||
|
if(c.equals(client)){
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
c.send(packet);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Отправить пакет всем клиентам с публичными ключами из списка publicKeys
|
* Отправить пакет всем клиентам с публичными ключами из списка publicKeys
|
||||||
* @param publicKeys список публичных ключей получателей
|
* @param publicKeys список публичных ключей получателей
|
||||||
|
|||||||
@@ -39,12 +39,9 @@ public class GroupRepository extends Repository<Group> {
|
|||||||
* @param groupId ID группы
|
* @param groupId ID группы
|
||||||
* @param creatorPublicKey публичный ключ создателя группы, который будет единственным участником группы при создании
|
* @param creatorPublicKey публичный ключ создателя группы, который будет единственным участником группы при создании
|
||||||
*/
|
*/
|
||||||
public void createGroup(String groupId, String creatorPublicKey) {
|
public void createGroup(String groupId) {
|
||||||
Group group = new Group();
|
Group group = new Group();
|
||||||
group.setGroupId(groupId);
|
group.setGroupId(groupId);
|
||||||
List<String> membersPublicKeys = new ArrayList<>();
|
|
||||||
membersPublicKeys.add(creatorPublicKey);
|
|
||||||
group.setMembersPublicKeys(membersPublicKeys);
|
|
||||||
this.save(group);
|
this.save(group);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ public class Executor17GroupCreate extends PacketExecutor<Packet17GroupCreate> {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
String groupId = RandomUtil.randomString(16);
|
String groupId = RandomUtil.randomString(16);
|
||||||
this.groupRepository.createGroup(groupId, eciAuthentificate.getPublicKey());
|
this.groupRepository.createGroup(groupId);
|
||||||
/**
|
/**
|
||||||
* Отправляем клиенту ид созданной группы
|
* Отправляем клиенту ид созданной группы
|
||||||
*/
|
*/
|
||||||
|
|||||||
@@ -3,17 +3,25 @@ package im.rosetta.executors;
|
|||||||
import im.rosetta.Failures;
|
import im.rosetta.Failures;
|
||||||
import im.rosetta.client.tags.ECIAuthentificate;
|
import im.rosetta.client.tags.ECIAuthentificate;
|
||||||
import im.rosetta.database.entity.Group;
|
import im.rosetta.database.entity.Group;
|
||||||
|
import im.rosetta.database.repository.BufferRepository;
|
||||||
import im.rosetta.database.repository.GroupRepository;
|
import im.rosetta.database.repository.GroupRepository;
|
||||||
import im.rosetta.packet.Packet20GroupJoin;
|
import im.rosetta.packet.Packet20GroupJoin;
|
||||||
import im.rosetta.packet.runtime.NetworkGroupStatus;
|
import im.rosetta.packet.runtime.NetworkGroupStatus;
|
||||||
|
import im.rosetta.service.services.BufferService;
|
||||||
import io.orprotocol.ProtocolException;
|
import io.orprotocol.ProtocolException;
|
||||||
import io.orprotocol.client.Client;
|
import io.orprotocol.client.Client;
|
||||||
import io.orprotocol.packet.PacketExecutor;
|
import io.orprotocol.packet.PacketExecutor;
|
||||||
|
import io.orprotocol.packet.PacketManager;
|
||||||
|
|
||||||
public class Executor20GroupJoin extends PacketExecutor<Packet20GroupJoin> {
|
public class Executor20GroupJoin extends PacketExecutor<Packet20GroupJoin> {
|
||||||
|
|
||||||
private final GroupRepository groupRepository = new GroupRepository();
|
private final GroupRepository groupRepository = new GroupRepository();
|
||||||
|
private final BufferRepository bufferRepository = new BufferRepository();
|
||||||
|
private final BufferService bufferService;
|
||||||
|
|
||||||
|
public Executor20GroupJoin(PacketManager packetManager) {
|
||||||
|
this.bufferService = new BufferService(bufferRepository, packetManager);
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void onPacketReceived(Packet20GroupJoin packet, Client client) throws Exception, ProtocolException {
|
public void onPacketReceived(Packet20GroupJoin packet, Client client) throws Exception, ProtocolException {
|
||||||
@@ -53,12 +61,21 @@ public class Executor20GroupJoin extends PacketExecutor<Packet20GroupJoin> {
|
|||||||
client.send(packet);
|
client.send(packet);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
/**
|
||||||
|
* Все проверки пройдены, клиент может вступить в группу
|
||||||
|
*/
|
||||||
|
packet.setStatus(NetworkGroupStatus.JOINED);
|
||||||
|
/**
|
||||||
|
* Кладем пакет в буфер для будущей синхронизации
|
||||||
|
*/
|
||||||
|
this.bufferService.pushPacketToBuffer("server", eciAuthentificate.getPublicKey(), packet);
|
||||||
/**
|
/**
|
||||||
* Добавляем клиента в группу и возвращаем клиенту статус JOINED
|
* Добавляем клиента в группу и возвращаем клиенту статус JOINED
|
||||||
*/
|
*/
|
||||||
this.groupRepository.addMemberToGroup(groupId, eciAuthentificate.getPublicKey());
|
this.groupRepository.addMemberToGroup(groupId, eciAuthentificate.getPublicKey());
|
||||||
packet.setStatus(NetworkGroupStatus.JOINED);
|
/**
|
||||||
|
* Возвращаем клиенту ответный пакет с статусом JOINED
|
||||||
|
*/
|
||||||
client.send(packet);
|
client.send(packet);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -14,11 +14,20 @@ public class Packet20GroupJoin extends Packet {
|
|||||||
|
|
||||||
private String groupId;
|
private String groupId;
|
||||||
private NetworkGroupStatus status;
|
private NetworkGroupStatus status;
|
||||||
|
/**
|
||||||
|
* Строка группы, которая содержит информацию о группе, такую как ее название, описание и ключ
|
||||||
|
* Строка зашифрована обратимым шифрованием, где ключом выступает - реальный приватный ключ
|
||||||
|
* входящего в группу клиента. Нужно это для будущей синхронзации, так как клиенту на его другом
|
||||||
|
* устройстве нужно получить ключ группы и ее информацию. Сервер расшифровать эту строку не может. Эту
|
||||||
|
* строку может расшифровать только клиент, так как она зашифрована его приватным ключом
|
||||||
|
*/
|
||||||
|
private String groupString;
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void read(Stream stream) {
|
public void read(Stream stream) {
|
||||||
this.groupId = stream.readString();
|
this.groupId = stream.readString();
|
||||||
this.status = NetworkGroupStatus.fromCode(stream.readInt8());
|
this.status = NetworkGroupStatus.fromCode(stream.readInt8());
|
||||||
|
this.groupString = stream.readString();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@@ -27,6 +36,7 @@ public class Packet20GroupJoin extends Packet {
|
|||||||
stream.writeInt16(this.packetId);
|
stream.writeInt16(this.packetId);
|
||||||
stream.writeString(this.groupId);
|
stream.writeString(this.groupId);
|
||||||
stream.writeInt8(this.status.getCode());
|
stream.writeInt8(this.status.getCode());
|
||||||
|
stream.writeString(this.groupString);
|
||||||
return stream;
|
return stream;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -62,4 +72,20 @@ public class Packet20GroupJoin extends Packet {
|
|||||||
this.status = status;
|
this.status = status;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Получить строку группы, которая содержит информацию о группе, такую как ее название, описание и ключ
|
||||||
|
* @return строка группы
|
||||||
|
*/
|
||||||
|
public String getGroupString() {
|
||||||
|
return groupString;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Установить строку группы, которая содержит информацию о группе, такую как ее название, описание и ключ
|
||||||
|
* @param groupString строка группы
|
||||||
|
*/
|
||||||
|
public void setGroupString(String groupString) {
|
||||||
|
this.groupString = groupString;
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,5 @@
|
|||||||
package im.rosetta.service.dispatch;
|
package im.rosetta.service.dispatch;
|
||||||
|
|
||||||
import java.util.HashSet;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
|
||||||
import im.rosetta.client.ClientManager;
|
import im.rosetta.client.ClientManager;
|
||||||
@@ -57,7 +56,21 @@ public class MessageDispatcher {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
/**
|
/**
|
||||||
* Отправляем всем участникам группы, кроме отправителя этот пакет, попутно не забывая проверить, а не один ли он в группе
|
* Ретранслируем сообщение ВСЕМ авторизованным сессиям отправителя КРОМЕ текущей,
|
||||||
|
* чтобы синхронизировать отправленные сообщения
|
||||||
|
*/
|
||||||
|
if(!(packet instanceof Packet11Typeing)){
|
||||||
|
/**
|
||||||
|
* Если это пакет печати его не обязательно кэшировать, так как он нужен только
|
||||||
|
* для отображения статуса печати в реальном времени
|
||||||
|
* Кладем пакет в буфер для будущей синхронизации и на случай если кто-то из участников оффлайн,
|
||||||
|
* в toPublicKey при отправке в группу у нас находится #group:groupId
|
||||||
|
*/
|
||||||
|
this.bufferService.pushPacketToBuffer(fromPublicKey, toPublicKey.replace("#group:", ""), packet);
|
||||||
|
this.clientManager.retranslate(client, packet);
|
||||||
|
}
|
||||||
|
/**
|
||||||
|
* Отправляем всем участникам группы, кроме отправителя, попутно проверяем, а не один ли он в группе
|
||||||
*/
|
*/
|
||||||
groupMembersPublicKeys.remove(eciAuthentificate.getPublicKey());
|
groupMembersPublicKeys.remove(eciAuthentificate.getPublicKey());
|
||||||
if(groupMembersPublicKeys.isEmpty()){
|
if(groupMembersPublicKeys.isEmpty()){
|
||||||
@@ -67,20 +80,10 @@ public class MessageDispatcher {
|
|||||||
*/
|
*/
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
/**
|
||||||
|
* Отправляем сообщение всем, кто в беседе
|
||||||
|
*/
|
||||||
this.clientManager.sendPacketToAuthorizedPK(groupMembersPublicKeys, packet);
|
this.clientManager.sendPacketToAuthorizedPK(groupMembersPublicKeys, packet);
|
||||||
|
|
||||||
if(packet instanceof Packet11Typeing){
|
|
||||||
/**
|
|
||||||
* Если это пакет печати его не обязательно кэшировать, так как он нужен только
|
|
||||||
* для отображения статуса печати в реальном времени
|
|
||||||
*/
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
/**
|
|
||||||
* Кладем пакет в буфер для будущей синхронизации и на случай если кто-то из участников оффлайн,
|
|
||||||
* в toPublicKey при отправке в группу у нас находится #group:groupId
|
|
||||||
*/
|
|
||||||
this.bufferService.pushPacketToBuffer(fromPublicKey, toPublicKey.replace("#group:", ""), packet);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -94,6 +97,12 @@ public class MessageDispatcher {
|
|||||||
public void sendPeer(PacketBaseDialog packet, Client client, boolean bufferizationNeed) throws ProtocolException {
|
public void sendPeer(PacketBaseDialog packet, Client client, boolean bufferizationNeed) throws ProtocolException {
|
||||||
String fromPublicKey = packet.getFromPublicKey();
|
String fromPublicKey = packet.getFromPublicKey();
|
||||||
String toPublicKey = packet.getToPublicKey();
|
String toPublicKey = packet.getToPublicKey();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Ретранслируем сообщение ВСЕМ авторизованным сессиям отправителя КРОМЕ текущей,
|
||||||
|
* чтобы синхронизировать отправленные сообщения
|
||||||
|
*/
|
||||||
|
this.clientManager.retranslate(client, packet);
|
||||||
this.clientManager.sendPacketToAuthorizedPK(toPublicKey, packet);
|
this.clientManager.sendPacketToAuthorizedPK(toPublicKey, packet);
|
||||||
|
|
||||||
if(!bufferizationNeed){
|
if(!bufferizationNeed){
|
||||||
@@ -107,11 +116,6 @@ public class MessageDispatcher {
|
|||||||
* Сохраняем сообщение в буфер на случай если получатель офлайн, или нам нужна будет синхронизация сообщений для получателя
|
* Сохраняем сообщение в буфер на случай если получатель офлайн, или нам нужна будет синхронизация сообщений для получателя
|
||||||
*/
|
*/
|
||||||
this.bufferService.pushPacketToBuffer(fromPublicKey, toPublicKey, packet);
|
this.bufferService.pushPacketToBuffer(fromPublicKey, toPublicKey, packet);
|
||||||
|
|
||||||
/**
|
|
||||||
* Ретранслируем сообщение всем авторизованным сессиям отправителя, чтобы синхронизировать отправленные сообщения
|
|
||||||
*/
|
|
||||||
this.retranslate(packet, client);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -127,34 +131,4 @@ public class MessageDispatcher {
|
|||||||
this.sendPeer(packet, client, true);
|
this.sendPeer(packet, client, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Сообщает всем авторизованным сессиям отправителя о том, что он отправил сообщения,
|
|
||||||
* для того чтобы синхронизировать отправленные сообщения на всех устройствах отправителя
|
|
||||||
* @param packet пакет сообщения
|
|
||||||
* @param client клиент отправляющий пакет
|
|
||||||
* @throws ProtocolException
|
|
||||||
*/
|
|
||||||
public void retranslate(PacketBaseDialog packet, Client client) throws ProtocolException {
|
|
||||||
ECIAuthentificate eciAuthentificate = client.getTag(ECIAuthentificate.class);
|
|
||||||
HashSet<Client> clients = this.clientManager.getClientIndexer()
|
|
||||||
.getClients(ECIAuthentificate.class, "publicKey", eciAuthentificate.getPublicKey());
|
|
||||||
if(clients == null){
|
|
||||||
/**
|
|
||||||
* Нет авторизованных сессий с таким публичным ключом
|
|
||||||
*/
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
for(Client c : clients){
|
|
||||||
/**
|
|
||||||
* Проходим по всем устройствам с таким публичным ключом и ретранслируем им пакет, кроме того устройства что
|
|
||||||
* отправило пакет
|
|
||||||
*/
|
|
||||||
if(c.equals(client)){
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
c.send(packet);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user