SHA256
Старт
This commit is contained in:
AidarKC
@@ -0,0 +1,66 @@
|
||||
package server.logic;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import server.logic.ws_protocol.binary.handlers.*;
|
||||
import server.logic.ws_protocol.WireCodes;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Обработчик входящих сообщение на сервер.
|
||||
* По коду сообщения (первые 4 байта сообщения) находи нужный хэндлер и передаёт в него сообщение
|
||||
* Получает и возвращает ответ от хэндлера
|
||||
*/
|
||||
public final class InboundMessageProcessor {
|
||||
private static final Logger log = LoggerFactory.getLogger(InboundMessageProcessor.class);
|
||||
|
||||
private static final Map<Integer, MessageHandler> HANDLERS = Map.of(
|
||||
WireCodes.Op.PING, new PingHandler(),
|
||||
WireCodes.Op.ADD_BLOCK, new AddBlockHandler(),
|
||||
WireCodes.Op.GET_BLOCKCHAIN,new GetBlockchainHandler(),
|
||||
WireCodes.Op.SEARCH_USERS, new SearchUsersHandler(),
|
||||
WireCodes.Op.GET_LAST_BLOCK_INFO,new GetLastBlockInfoHandler()
|
||||
|
||||
);
|
||||
|
||||
private InboundMessageProcessor() {}
|
||||
|
||||
public static byte[] process(byte[] msg) {
|
||||
if (msg == null || msg.length < 4)
|
||||
return intTo4Bytes(WireCodes.Status.BAD_REQUEST);
|
||||
|
||||
int op = first4ToInt(msg);
|
||||
MessageHandler h = HANDLERS.get(op);
|
||||
if (h == null) {
|
||||
log.warn("Неизвестная операция: {}", op);
|
||||
return intTo4Bytes(WireCodes.Status.BAD_REQUEST);
|
||||
}
|
||||
|
||||
try {
|
||||
return h.handle(msg);
|
||||
} catch (Exception e) {
|
||||
log.error("Ошибка при обработке операции {}", op, e);
|
||||
return intTo4Bytes(WireCodes.Status.INTERNAL_ERROR);
|
||||
}
|
||||
}
|
||||
|
||||
private static int first4ToInt(byte[] msg) {
|
||||
return ByteBuffer.wrap(msg, 0, 4)
|
||||
.order(ByteOrder.BIG_ENDIAN)
|
||||
.getInt();
|
||||
}
|
||||
|
||||
public static byte[] intTo4Bytes(int code) {
|
||||
return ByteBuffer.allocate(4)
|
||||
.order(ByteOrder.BIG_ENDIAN)
|
||||
.putInt(code)
|
||||
.array();
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,250 @@
|
||||
package server.logic.ws_protocol.binary.handlers;
|
||||
|
||||
import blockchain.BchBlockEntry;
|
||||
import blockchain.body.BodyRecord;
|
||||
import blockchain.BodyRecordParser;
|
||||
import blockchain.body.HeaderBody;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import server.logic.ws_protocol.WireCodes;
|
||||
import utils.blockchain.BchInfoEntry;
|
||||
import utils.blockchain.BchInfoManager;
|
||||
import utils.crypto.BchCryptoVerifier;
|
||||
import utils.files.FileStoreUtil;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
import java.util.Arrays;
|
||||
|
||||
/**
|
||||
* AddBlockHandler — обработчик команды "добавить блок" (ADD_BLOCK)
|
||||
* ---------------------------------------------------------------
|
||||
* Принимает бинарное сообщение от клиента и добавляет новый блок в цепочку.
|
||||
*
|
||||
* Формат входного сообщения (msg):
|
||||
* [0..3] — 4 байта: код операции (WireCodes.ADD_BLOCK)
|
||||
* [4..11] — 8 байт: blockchainId (уникальный идентификатор цепочки)
|
||||
* [12..] — байты полного блока .bch:
|
||||
* ├── 4 байта recordSize = M + 18
|
||||
* ├── 4 байта recordNumber
|
||||
* ├── 8 байт timestamp
|
||||
* ├── 2 байта recordType
|
||||
* ├── 2 байта recordVersion
|
||||
* ├── M байт body (содержимое блока)
|
||||
* ├── 64 байта signature (Ed25519)
|
||||
* └── 32 байта hash (SHA-256)
|
||||
*
|
||||
* ---------------------------------------------------------------
|
||||
* Алгоритм работы:
|
||||
*
|
||||
* 1️⃣ Распаковать BchBlockEntry из msg (т.е. выделить тело блока и подписи).
|
||||
* 2️⃣ Найти описание цепочки (BchInfoEntry) по blockchainId.
|
||||
*
|
||||
* ─ Если описания нет (цепочка ещё не существует):
|
||||
* • принимаем только блок типа 0 (HeaderBody) и номера 0;
|
||||
* • парсим его, создаём новый BchInfoEntry на основе данных заголовка;
|
||||
* • проверяем подпись и хэш;
|
||||
* • проверяем корректность тела блока (check);
|
||||
* • сохраняем блок и создаём новый blockchain-файл;
|
||||
* • добавляем цепочку в менеджер BchInfoManager.
|
||||
* (💡 временное решение: создание цепочки допустимо только через HeaderBody)
|
||||
*
|
||||
* ─ Если цепочка уже существует:
|
||||
* • проверяем, что номер блока равен (lastBlockNumber + 1);
|
||||
* • проверяем подпись и хэш;
|
||||
* • проверяем тело блока (check);
|
||||
* • добавляем блок в файл цепочки;
|
||||
* • обновляем состояние BchInfoEntry (номер, хэш, размер).
|
||||
*
|
||||
* 3️⃣ Если все проверки пройдены — возвращаем статус OK.
|
||||
*
|
||||
* Таким образом, единственное различие между первым блоком и последующими —
|
||||
* момент инициализации описания цепочки (BchInfoEntry).
|
||||
* Всё остальное (валидация, подпись, добавление, обновление) выполняется одинаково.
|
||||
*/
|
||||
public class AddBlockHandler implements MessageHandler {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(AddBlockHandler.class);
|
||||
|
||||
@Override
|
||||
public byte[] handle(byte[] msg) {
|
||||
try {
|
||||
// =====================================================================
|
||||
// 1️⃣ Проверка минимальной длины пакета
|
||||
// =====================================================================
|
||||
int minFull = BchBlockEntry.RAW_HEADER_SIZE + BchBlockEntry.SIGNATURE_LEN + BchBlockEntry.HASH_LEN;
|
||||
// (RAW_HEADER_SIZE = 18 байт, подпись = 64, хэш = 32)
|
||||
if (msg.length < 4 + 8 + minFull)
|
||||
return code(WireCodes.Status.BAD_REQUEST);
|
||||
|
||||
// =====================================================================
|
||||
// 2️⃣ Извлекаем blockchainId (8 байт начиная с позиции 4)
|
||||
// =====================================================================
|
||||
long blockchainId = ByteBuffer.wrap(msg, 4, 8)
|
||||
.order(ByteOrder.BIG_ENDIAN)
|
||||
.getLong();
|
||||
|
||||
// Всё, что дальше, — это бинарное содержимое блока .bch
|
||||
int offset = 12; // первые 12 байт = код + blockchainId
|
||||
|
||||
// =====================================================================
|
||||
// 3️⃣ Парсим блок (RAW + подпись + хэш)
|
||||
// =====================================================================
|
||||
byte[] fullBlock = Arrays.copyOfRange(msg, offset, msg.length);
|
||||
BchBlockEntry block = new BchBlockEntry(fullBlock); // сам распакует RAW-часть и подписи
|
||||
|
||||
// =====================================================================
|
||||
// 4️⃣ Получаем текущее описание цепочки (BchInfoEntry)
|
||||
// =====================================================================
|
||||
BchInfoManager info = BchInfoManager.getInstance();
|
||||
BchInfoEntry chain = info.getBchInfo(blockchainId);
|
||||
|
||||
byte[] prevHash32;
|
||||
int expectedNum;
|
||||
String userLogin;
|
||||
byte[] publicKey32;
|
||||
|
||||
// =====================================================================
|
||||
// 🧩 СЦЕНАРИЙ 1: цепочка отсутствует — создаём новую
|
||||
// =====================================================================
|
||||
if (chain == null) {
|
||||
// Допускаем только блок-заголовок (type=0, num=0)
|
||||
if (block.recordType != BchBlockEntry.TYPE_HEADER || block.recordNumber != 0) {
|
||||
log.warn("Попытка создать новую цепочку без корректного заголовка (type={}, num={})",
|
||||
block.recordType, block.recordNumber);
|
||||
return code(WireCodes.Status.BAD_REQUEST);
|
||||
}
|
||||
|
||||
// Парсим тело блока → HeaderBody
|
||||
BodyRecord body = BodyRecordParser.parse(block.recordType, block.recordTypeVersion, block.body).check();
|
||||
if (!(body instanceof HeaderBody))
|
||||
return code(WireCodes.Status.BAD_REQUEST);
|
||||
|
||||
HeaderBody hb = (HeaderBody) body;
|
||||
|
||||
// Проверяем, что blockchainId совпадает
|
||||
if (hb.blockchainId != blockchainId) {
|
||||
log.warn("Несовпадение blockchainId в заголовке (ожидалось {}, получено {})",
|
||||
blockchainId, hb.blockchainId);
|
||||
return code(WireCodes.Status.BAD_REQUEST);
|
||||
}
|
||||
|
||||
// Проверяем подпись и хэш первого блока (предыдущий хэш = 0)
|
||||
prevHash32 = new byte[32];
|
||||
boolean verified = BchCryptoVerifier.verifyAll(
|
||||
hb.userLogin,
|
||||
blockchainId,
|
||||
prevHash32,
|
||||
block.rawBytes,
|
||||
block.getSignature64(),
|
||||
block.getHash32(),
|
||||
hb.publicKey32
|
||||
);
|
||||
if (!verified) {
|
||||
log.warn("❌ Подпись не прошла проверку при создании цепочки blockchainId={}", blockchainId);
|
||||
return code(WireCodes.Status.UNVERIFIED);
|
||||
}
|
||||
|
||||
// ✅ Всё хорошо: создаём новую цепочку
|
||||
info.addBlockchain(blockchainId, hb.userLogin, hb.publicKey32, Integer.MAX_VALUE);
|
||||
info.updateBlockchainState(blockchainId, block.recordNumber, bytesToHex(block.getHash32()), fullBlock.length);
|
||||
|
||||
FileStoreUtil.getInstance().addDataToBlockchain(blockchainId, fullBlock);
|
||||
|
||||
log.info("✅ Создана новая цепочка blockchainId={}, user={}, blockNum={}",
|
||||
blockchainId, hb.userLogin, block.recordNumber);
|
||||
|
||||
return code(WireCodes.Status.OK);
|
||||
}
|
||||
|
||||
// =====================================================================
|
||||
// 🧩 СЦЕНАРИЙ 2: цепочка существует — добавляем новый блок
|
||||
// =====================================================================
|
||||
expectedNum = chain.lastBlockNumber + 1;
|
||||
|
||||
// Проверка последовательности (и отправка lastBlockNumber)
|
||||
if (block.recordNumber < expectedNum) {
|
||||
log.info("🔁 Блок {} уже существует, последний = {}", block.recordNumber, chain.lastBlockNumber);
|
||||
ByteBuffer out = ByteBuffer.allocate(8).order(ByteOrder.BIG_ENDIAN);
|
||||
out.putInt(WireCodes.Status.BLOCK_ALREADY_EXISTS);
|
||||
out.putInt(chain.lastBlockNumber);
|
||||
return out.array();
|
||||
}
|
||||
if (block.recordNumber > expectedNum) {
|
||||
log.warn("⚠️ Нарушена последовательность: получен {}, ожидался {}", block.recordNumber, expectedNum);
|
||||
ByteBuffer out = ByteBuffer.allocate(8).order(ByteOrder.BIG_ENDIAN);
|
||||
out.putInt(WireCodes.Status.OUT_OF_SEQUENCE);
|
||||
out.putInt(chain.lastBlockNumber);
|
||||
return out.array();
|
||||
}
|
||||
|
||||
userLogin = chain.userLogin;
|
||||
publicKey32 = chain.getPublicKey32();
|
||||
|
||||
// Хэш предыдущего блока (или 32 нуля, если это первый)
|
||||
prevHash32 = (chain.lastBlockHash == null || chain.lastBlockHash.isEmpty())
|
||||
? new byte[32]
|
||||
: hexToBytes(chain.lastBlockHash);
|
||||
|
||||
// Проверяем подпись и хэш
|
||||
boolean verified = BchCryptoVerifier.verifyAll(
|
||||
userLogin,
|
||||
blockchainId,
|
||||
prevHash32,
|
||||
block.rawBytes,
|
||||
block.getSignature64(),
|
||||
block.getHash32(),
|
||||
publicKey32
|
||||
);
|
||||
if (!verified) {
|
||||
log.warn("❌ Подпись не прошла проверку: chainId={}, blockNum={}", blockchainId, block.recordNumber);
|
||||
return code(WireCodes.Status.UNVERIFIED);
|
||||
}
|
||||
|
||||
// Проверяем тело блока (например, корректный UTF-8 или структура)
|
||||
BodyRecord body = BodyRecordParser.parse(block.recordType, block.recordTypeVersion, block.body).check();
|
||||
|
||||
// ✅ Добавляем блок в файл цепочки
|
||||
FileStoreUtil.getInstance().addDataToBlockchain(blockchainId, fullBlock);
|
||||
|
||||
// Обновляем состояние цепочки (номер, хэш, размер)
|
||||
int newSize = chain.blockchainSize + fullBlock.length;
|
||||
info.updateBlockchainState(blockchainId, block.recordNumber, bytesToHex(block.getHash32()), newSize);
|
||||
|
||||
log.info("✅ Блок добавлен: chain={}, num={}, type={}, bytes={}",
|
||||
blockchainId, block.recordNumber, block.recordType, fullBlock.length);
|
||||
|
||||
return code(WireCodes.Status.OK);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("❌ ADD_BLOCK: внутренняя ошибка при обработке", e);
|
||||
return code(WireCodes.Status.INTERNAL_ERROR);
|
||||
}
|
||||
}
|
||||
|
||||
// =====================================================================
|
||||
// Утилиты
|
||||
// =====================================================================
|
||||
|
||||
/** Преобразовать статус (int) в 4 байта BigEndian. */
|
||||
private static byte[] code(int status) {
|
||||
return ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putInt(status).array();
|
||||
}
|
||||
|
||||
/** Конвертация HEX → bytes (для хэшей). */
|
||||
private static byte[] hexToBytes(String hex) {
|
||||
int len = hex.length();
|
||||
byte[] out = new byte[len / 2];
|
||||
for (int i = 0; i < len; i += 2)
|
||||
out[i / 2] = (byte) Integer.parseInt(hex.substring(i, i + 2), 16);
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Конвертация bytes → HEX (для сохранения в BchInfo). */
|
||||
private static String bytesToHex(byte[] b) {
|
||||
StringBuilder sb = new StringBuilder(b.length * 2);
|
||||
for (byte x : b) sb.append(String.format("%02x", x));
|
||||
return sb.toString();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
package server.logic.ws_protocol.binary.handlers;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import server.logic.ws_protocol.WireCodes;
|
||||
import utils.files.FileStoreUtil;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
|
||||
/**
|
||||
* Возврат полного содержимого блокчейна (GET_BLOCKCHAIN).
|
||||
*/
|
||||
public class GetBlockchainHandler implements MessageHandler {
|
||||
private static final Logger log = LoggerFactory.getLogger(GetBlockchainHandler.class);
|
||||
|
||||
@Override
|
||||
public byte[] handle(byte[] msg) {
|
||||
try {
|
||||
if (msg.length < 12)
|
||||
return intTo4Bytes(WireCodes.Status.BAD_REQUEST);
|
||||
|
||||
long id = ByteBuffer.wrap(msg, 4, 8)
|
||||
.order(ByteOrder.BIG_ENDIAN)
|
||||
.getLong();
|
||||
|
||||
FileStoreUtil fs = FileStoreUtil.getInstance();
|
||||
byte[] data = fs.readAllDataFromBlockchain(id);
|
||||
|
||||
return packOk(data);
|
||||
|
||||
} catch (IllegalStateException e) {
|
||||
log.warn("GET_BLOCKCHAIN: файл не найден ({})", e.getMessage());
|
||||
return intTo4Bytes(WireCodes.Status.CHAIN_NOT_FOUND);
|
||||
} catch (Exception e) {
|
||||
log.error("GET_BLOCKCHAIN: ошибка", e);
|
||||
return intTo4Bytes(WireCodes.Status.INTERNAL_ERROR);
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] packOk(byte[] data) {
|
||||
if (data == null) data = new byte[0];
|
||||
ByteBuffer out = ByteBuffer.allocate(8 + data.length).order(ByteOrder.BIG_ENDIAN);
|
||||
out.putInt(WireCodes.Status.OK);
|
||||
out.putInt(data.length);
|
||||
out.put(data);
|
||||
return out.array();
|
||||
}
|
||||
|
||||
private static byte[] intTo4Bytes(int code) {
|
||||
return ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putInt(code).array();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package server.logic.ws_protocol.binary.handlers;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import server.logic.ws_protocol.WireCodes;
|
||||
import utils.blockchain.BchInfoEntry;
|
||||
import utils.blockchain.BchInfoManager;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
import java.util.Arrays;
|
||||
|
||||
/**
|
||||
* Возврат информации о последнем блоке цепочки (GET_LAST_BLOCK_INFO).
|
||||
*/
|
||||
public class GetLastBlockInfoHandler implements MessageHandler {
|
||||
private static final Logger log = LoggerFactory.getLogger(GetLastBlockInfoHandler.class);
|
||||
|
||||
@Override
|
||||
public byte[] handle(byte[] msg) {
|
||||
try {
|
||||
if (msg.length < 12)
|
||||
return intTo4Bytes(WireCodes.Status.BAD_REQUEST);
|
||||
|
||||
long blockchainId = ByteBuffer.wrap(msg, 4, 8)
|
||||
.order(ByteOrder.BIG_ENDIAN)
|
||||
.getLong();
|
||||
|
||||
BchInfoManager mgr = BchInfoManager.getInstance();
|
||||
BchInfoEntry entry = mgr.getBchInfo(blockchainId);
|
||||
if (entry == null)
|
||||
return intTo4Bytes(WireCodes.Status.CHAIN_NOT_FOUND);
|
||||
|
||||
int lastNum = entry.lastBlockNumber;
|
||||
byte[] hash = hexToBytes(entry.lastBlockHash);
|
||||
|
||||
ByteBuffer out = ByteBuffer.allocate(4 + 4 + 32).order(ByteOrder.BIG_ENDIAN);
|
||||
out.putInt(WireCodes.Status.OK);
|
||||
out.putInt(lastNum);
|
||||
out.put(hash);
|
||||
return out.array();
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("GET_LAST_BLOCK_INFO: ошибка", e);
|
||||
return intTo4Bytes(WireCodes.Status.INTERNAL_ERROR);
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] intTo4Bytes(int code) {
|
||||
return ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putInt(code).array();
|
||||
}
|
||||
|
||||
private static byte[] hexToBytes(String hex) {
|
||||
if (hex == null || hex.isEmpty()) return new byte[32];
|
||||
int len = hex.length();
|
||||
byte[] out = new byte[len / 2];
|
||||
for (int i = 0; i < len; i += 2)
|
||||
out[i / 2] = (byte) Integer.parseInt(hex.substring(i, i + 2), 16);
|
||||
if (out.length < 32) { // добиваем нулями
|
||||
byte[] full = new byte[32];
|
||||
System.arraycopy(out, 0, full, 32 - out.length, out.length);
|
||||
return full;
|
||||
}
|
||||
return Arrays.copyOf(out, 32);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package server.logic.ws_protocol.binary.handlers;
|
||||
|
||||
/**
|
||||
* Общий интерфейс для всех обработчиков входящих сообщений.
|
||||
*/
|
||||
public interface MessageHandler {
|
||||
/**
|
||||
* Обработать входящее сообщение и вернуть бинарный ответ.
|
||||
*/
|
||||
byte[] handle(byte[] msg);
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package server.logic.ws_protocol.binary.handlers;
|
||||
|
||||
import server.logic.ws_protocol.WireCodes;
|
||||
|
||||
/**
|
||||
* Обработчик команды PING.
|
||||
* Возвращает просто статус PONG.
|
||||
*/
|
||||
public class PingHandler implements MessageHandler {
|
||||
@Override
|
||||
public byte[] handle(byte[] msg) {
|
||||
return new byte[]{
|
||||
0, 0, 0, (byte) WireCodes.Status.PONG // проще и быстрее, можно и через ByteBuffer
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package server.logic.ws_protocol.binary.handlers;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import server.logic.ws_protocol.WireCodes;
|
||||
import utils.search.UserSearchService;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Поиск пользователей по логину (SEARCH_USERS).
|
||||
*/
|
||||
public class SearchUsersHandler implements MessageHandler {
|
||||
private static final Logger log = LoggerFactory.getLogger(SearchUsersHandler.class);
|
||||
|
||||
@Override
|
||||
public byte[] handle(byte[] msg) {
|
||||
try {
|
||||
if (msg.length < 8)
|
||||
return intTo4Bytes(WireCodes.Status.BAD_REQUEST);
|
||||
|
||||
int N = ByteBuffer.wrap(msg, 4, 4).order(ByteOrder.BIG_ENDIAN).getInt();
|
||||
if (N < 0 || msg.length < 8 + N)
|
||||
return intTo4Bytes(WireCodes.Status.BAD_REQUEST);
|
||||
|
||||
String query = new String(msg, 8, N, StandardCharsets.UTF_8);
|
||||
List<UserSearchService.Pair> found = UserSearchService.getInstance().searchFirst5(query);
|
||||
return pack(found);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("SEARCH_USERS: ошибка", e);
|
||||
return intTo4Bytes(WireCodes.Status.INTERNAL_ERROR);
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] pack(List<UserSearchService.Pair> pairs) {
|
||||
if (pairs == null) pairs = List.of();
|
||||
int total = 8;
|
||||
var chunks = new java.util.ArrayList<byte[]>();
|
||||
for (var p : pairs) {
|
||||
byte[] packed = UserSearchService.packPair(p);
|
||||
chunks.add(packed);
|
||||
total += packed.length;
|
||||
}
|
||||
|
||||
ByteBuffer out = ByteBuffer.allocate(total).order(ByteOrder.BIG_ENDIAN);
|
||||
out.putInt(WireCodes.Status.OK);
|
||||
out.putInt(pairs.size());
|
||||
for (var c : chunks) out.put(c);
|
||||
return out.array();
|
||||
}
|
||||
|
||||
private static byte[] intTo4Bytes(int code) {
|
||||
return ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putInt(code).array();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
package server.ws;
|
||||
|
||||
import org.eclipse.jetty.websocket.api.Session;
|
||||
import org.eclipse.jetty.websocket.api.WriteCallback;
|
||||
import org.eclipse.jetty.websocket.api.annotations.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import server.logic.InboundMessageProcessor;
|
||||
import server.logic.ws_protocol.JSON.ConnectionContext;
|
||||
import server.logic.ws_protocol.JSON.JsonInboundProcessor;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
@WebSocket
|
||||
public class BlockchainWsEndpoint {
|
||||
private static final Logger log = LoggerFactory.getLogger(BlockchainWsEndpoint.class);
|
||||
|
||||
private Session session;
|
||||
|
||||
/** Контекст для текущего WebSocket-соединения. */
|
||||
private final ConnectionContext connectionContext = new ConnectionContext();
|
||||
|
||||
@OnWebSocketConnect
|
||||
public void onConnect(Session session) {
|
||||
this.session = session;
|
||||
log.info("WS connected: {}", session.getRemoteAddress());
|
||||
}
|
||||
|
||||
@OnWebSocketMessage
|
||||
public void onBinary(byte[] payload, int offset, int length) {
|
||||
byte[] msg = new byte[length];
|
||||
System.arraycopy(payload, offset, msg, 0, length);
|
||||
|
||||
// Асинхронно обрабатываем входящее бинарное сообщение
|
||||
CompletableFuture
|
||||
.supplyAsync(() -> InboundMessageProcessor.process(msg))
|
||||
.thenAccept(resp -> {
|
||||
if (resp != null && session != null && session.isOpen()) {
|
||||
session.getRemote().sendBytes(ByteBuffer.wrap(resp), new WriteCallback() {
|
||||
@Override
|
||||
public void writeFailed(Throwable x) {
|
||||
log.warn("Failed to send response", x);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeSuccess() {
|
||||
log.debug("Response sent successfully");
|
||||
}
|
||||
});
|
||||
}
|
||||
})
|
||||
.exceptionally(ex -> {
|
||||
log.error("Processing failed", ex);
|
||||
trySendCode(500);
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
private void trySendCode(int code) {
|
||||
if (session != null && session.isOpen()) {
|
||||
byte[] resp = InboundMessageProcessor.intTo4Bytes(code);
|
||||
session.getRemote().sendBytes(ByteBuffer.wrap(resp), new WriteCallback() {
|
||||
@Override
|
||||
public void writeFailed(Throwable x) {
|
||||
log.warn("Failed to send error code", x);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeSuccess() {
|
||||
log.debug("Error code {} sent", code);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@OnWebSocketClose
|
||||
public void onClose(int statusCode, String reason) {
|
||||
log.info("WS closed: {} {}", statusCode, reason);
|
||||
// На всякий случай очищаем контекст
|
||||
connectionContext.reset();
|
||||
}
|
||||
|
||||
@OnWebSocketError
|
||||
public void onError(Throwable cause) {
|
||||
log.error("WS error", cause);
|
||||
}
|
||||
|
||||
// Обработка текстовых JSON-запросов
|
||||
@OnWebSocketMessage
|
||||
public void onText(String message) {
|
||||
log.info("📥 Получено TEXT-сообщение от клиента: {}", message);
|
||||
|
||||
CompletableFuture
|
||||
.supplyAsync(() -> JsonInboundProcessor.processJson(message, connectionContext))
|
||||
.thenAccept(respJson -> {
|
||||
if (respJson != null && session != null && session.isOpen()) {
|
||||
|
||||
log.info("📤 Отправляем ответ клиенту: {}", respJson);
|
||||
|
||||
session.getRemote().sendString(respJson, new WriteCallback() {
|
||||
@Override
|
||||
public void writeFailed(Throwable x) {
|
||||
log.warn("⚠️ Не удалось отправить JSON-ответ клиенту: {}", x.toString());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeSuccess() {
|
||||
log.debug("✔ JSON-ответ успешно отправлен");
|
||||
}
|
||||
});
|
||||
}
|
||||
})
|
||||
.exceptionally(ex -> {
|
||||
log.error("❌ Ошибка при обработке JSON-сообщения", ex);
|
||||
trySendJsonError();
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
private void trySendJsonError() {
|
||||
if (session != null && session.isOpen()) {
|
||||
String resp = "{\"op\":null,\"requestId\":null,\"status\":500,"
|
||||
+ "\"payload\":{\"code\":\"INTERNAL_ERROR\",\"message\":\"Ошибка сервера\"}}";
|
||||
|
||||
log.info("📤 Отправляем клиенту ошибку JSON: {}", resp);
|
||||
|
||||
session.getRemote().sendString(resp, new WriteCallback() {
|
||||
@Override
|
||||
public void writeFailed(Throwable x) {
|
||||
log.warn("⚠️ Не удалось отправить JSON-ответ клиенту: {}", x.toString());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeSuccess() {
|
||||
log.debug("✔ JSON-ошибка успешно отправлена");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
# server.ws
|
||||
|
||||
Пакет `server.ws` отвечает за сетевой уровень: WebSocket-сервер и обработку соединений.
|
||||
|
||||
Он принимает бинарные сообщения от клиентов, передаёт их в логику сервера и отправляет бинарные ответы обратно.
|
||||
|
||||
---
|
||||
|
||||
## Классы
|
||||
|
||||
### 1. `BlockchainWsEndpoint`
|
||||
WebSocket-эндпоинт для одного соединения.
|
||||
|
||||
Роль:
|
||||
- держит сессию с конкретным клиентом,
|
||||
- принимает сообщения,
|
||||
- вызывает бизнес-логику,
|
||||
- отправляет ответ.
|
||||
|
||||
Публичные методы (Jetty WebSocket lifecycle):
|
||||
|
||||
- `onConnect(Session session)`
|
||||
Вызывается Jetty при подключении клиента.
|
||||
Сохраняет `session`, пишет лог.
|
||||
|
||||
- `onBinary(byte[] payload, int offset, int length)`
|
||||
Клиент прислал бинарные данные.
|
||||
Логика:
|
||||
1. Копируем полезные байты.
|
||||
2. Передаём их в `InboundMessageProcessor.process(...)`.
|
||||
3. Асинхронно отправляем ответ обратно через `session.getRemote().sendBytes(...)`.
|
||||
|
||||
Ответ сервера — это либо `[4]statusCode`, либо `[4]OK + ...payload...` (в зависимости от операции).
|
||||
|
||||
- `onClose(int statusCode, String reason)`
|
||||
Логируем закрытие сессии.
|
||||
|
||||
- `onError(Throwable cause)`
|
||||
Логируем ошибку.
|
||||
|
||||
Внутренние (служебные):
|
||||
- `trySendCode(int code)` — отправить просто код ошибки, если что-то пошло не так.
|
||||
|
||||
Замечание: сам `BlockchainWsEndpoint` не знает протокола. Он просто прокидывает байты в `InboundMessageProcessor`.
|
||||
|
||||
---
|
||||
|
||||
### 2. `WsServer`
|
||||
Отдельный класс-ланчер. Поднимает Jetty WebSocket сервер.
|
||||
|
||||
Роль:
|
||||
- стартует HTTP-сервер Jetty на порту `8080`,
|
||||
- вешает WebSocket endpoint `/ws`,
|
||||
- задаёт таймаут бездействия.
|
||||
|
||||
Публичный метод:
|
||||
- `public static void main(String[] args)`
|
||||
Запуск сервера. Делает:
|
||||
- `new Server(8080)`
|
||||
- создаёт `ServletContextHandler`
|
||||
- через `JettyWebSocketServletContainerInitializer.configure(...)` регистрирует маппинг `/ws` → `BlockchainWsEndpoint`
|
||||
- `server.start(); server.join();`
|
||||
|
||||
После запуска сервер слушает `ws://localhost:8080/ws`.
|
||||
|
||||
---
|
||||
|
||||
## Как это стыкуется с остальной системой
|
||||
|
||||
1. Клиент открывает WebSocket на `/ws`.
|
||||
2. Шлёт бинарный пакет: `[4 байта opCode][дальше payload]`.
|
||||
3. `BlockchainWsEndpoint.onBinary()` → `InboundMessageProcessor.process(...)`.
|
||||
4. `InboundMessageProcessor` разбирает команду:
|
||||
- добавить блок
|
||||
- выдать блокчейн
|
||||
- поиск пользователей
|
||||
- ping
|
||||
5. Ответ упаковывается в бинарный формат и отправляется обратно через `BlockchainWsEndpoint`.
|
||||
|
||||
---
|
||||
|
||||
## Кратко
|
||||
|
||||
- `WsServer` = сервер, который слушает порт и вешает `/ws`.
|
||||
- `BlockchainWsEndpoint` = обработчик одного WebSocket-подключения, мост между сетью и логикой.
|
||||
@@ -0,0 +1,48 @@
|
||||
package server.ws;
|
||||
|
||||
import org.eclipse.jetty.server.Server;
|
||||
import org.eclipse.jetty.servlet.ServletContextHandler;
|
||||
import org.eclipse.jetty.websocket.server.config.JettyWebSocketServletContainerInitializer;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import shine.db.dao.SolanaUsersDAO;
|
||||
import shine.db.entities.SolanaUser;
|
||||
import utils.config.AppConfig;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* WsServer — поднимает Jetty WS на /ws (порт 8080).
|
||||
*/
|
||||
public final class WsServer {
|
||||
private static final Logger log = LoggerFactory.getLogger(WsServer.class);
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
AppConfig config = AppConfig.getInstance();
|
||||
int port = 7070;
|
||||
try {
|
||||
port = Integer.parseInt(config.getParam("server.port"),7070);
|
||||
} catch (Exception e) {
|
||||
log.info("Установите параметр server.port в файле настроек");
|
||||
}
|
||||
|
||||
Server server = new Server(port);
|
||||
|
||||
ServletContextHandler context = new ServletContextHandler();
|
||||
context.setContextPath("/");
|
||||
server.setHandler(context);
|
||||
|
||||
// Инициализация контейнера WebSocket
|
||||
JettyWebSocketServletContainerInitializer.configure(context, (servletContext, wsContainer) -> {
|
||||
// Таймаут простоя соединения (Jetty 11 синтаксис)
|
||||
wsContainer.setIdleTimeout(Duration.ofMinutes(5));
|
||||
|
||||
// Маппинг эндпоинта
|
||||
wsContainer.addMapping("/ws", (req, resp) -> new BlockchainWsEndpoint());
|
||||
});
|
||||
|
||||
server.start();
|
||||
log.info("✅ WS сервер запущен на ws://localhost:{}/ws", port);
|
||||
server.join();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user