Package xyz.cofe.trambda.tcp

Class TcpProtocol

java.lang.Object

xyz.cofe.trambda.tcp.TcpProtocol

public class TcpProtocol
extends Object

Поддержка TCP протокола.
Используется в клиенте TcpClient, взаимодействует с сессией TcpSession.
Умеет

• отправку сообщений Message, через методы send(Message, HeaderValue[]), send(Message, Consumer, HeaderValue[]), sendRaw(String, byte[], Consumer, HeaderValue[])
• читает очередное сообщение readNow()
• обрабатывает входящие сообщения process(Message, TcpHeader)

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	TcpProtocol.SentRawData

Field Summary

Fields

Modifier and Type	Field	Description
protected Supplier<InputStream>	getInput	Канал/поток входящих сообщений
protected Supplier<xyz.cofe.trambda.log.api.Logger>	getLogger	Логирование
protected Supplier<OutputStream>	getOutput	Канал/поток исходящих сообщений
protected ThreadLocal<Message>	sendMessage
Consumer<TcpProtocol.SentRawData>	sentRawDataConsumer
protected Map<String,Set<Consumer<ServerEvent>>>	serverEventListeners	Подписчики на серверные события
protected AtomicInteger	sid	Счетчик/генератор идентификаторов сообщений
protected Socket	socket	Сокет с которым производится работа

Constructor Summary

Constructors

Constructor	Description
TcpProtocol(OutputStream outputStream, InputStream inputStream, xyz.cofe.trambda.log.api.Logger logger)	Конструктор
TcpProtocol(OutputStream outputStream, InputStream inputStream, xyz.cofe.trambda.log.api.Logger logger, Consumer<TcpProtocol.SentRawData> sentRawDataConsumer)	Конструктор
TcpProtocol(Socket socket)	Конструктор
TcpProtocol(Socket socket, Consumer<TcpProtocol.SentRawData> sentRawDataConsumer)	Конструктор
TcpProtocol(Socket socket, xyz.cofe.trambda.log.api.Logger logger)	Конструктор

Method Summary

Modifier and Type	Method	Description
ResultConsumer<Compile,CompileResult>	compile(xyz.cofe.trambda.LambdaDump methodDef)	Компиляция лямбды
ResultConsumer<Execute,ExecuteResult>	execute(CompileResult cres)	Выполнение ранее скомпилированной лямбды
Map<Integer,Consumer<ErrMessage>>	getErrorConsumers()	Возвращает обработчики ошибок на конкретные запросы TcpHeader.getSid()
Map<Integer,Consumer<? extends Message>>	getResponseConsumers()	Возвращает обработчики на конкретные запросы TcpHeader.getSid()
AutoCloseable	listenServerEvent(String publisher, Consumer<ServerEvent> listener)	Добавляет подписчика на событие сервера.
int	ping(Consumer<Pong> consumer)	Отправляет Ping запрос
protected void	process(Message msg, TcpHeader header)	Обработка входящего сообщения.
protected void	process(Ping ping, TcpHeader header)	При получении Ping сообщения, отсылает ответ Pong
protected void	process(Pong pong, TcpHeader header)	При получении ответа Pong уведомляет подписанотов ping(Consumer), тех кто отправил запрос Ping
protected void	process(ServerEvent sevent, TcpHeader header)	Обработка входящего серверного события Уведомляет подписчиков listenServerEvent(String, Consumer) о серверном событии
boolean	readNow()	Читает сообщение Message из входного потока (getInput).
Optional<xyz.cofe.fn.Tuple2<RawPack,Optional<BuffInputStream>>>	receiveRaw(Optional<BuffInputStream> prevStream)	Получение сообщения из сети
void	removeServerEventListener(Consumer<ServerEvent> listener)	Отписка от событий сервера
int	send(Message message, Consumer<Integer> sid, HeaderValue<? extends Object>... headerValues)	Отправка сообщения
int	send(Message message, HeaderValue<? extends Object>... headerValues)	Отправка сообщения
int	sendRaw(String method, byte[] payload, Consumer<Integer> sendId, HeaderValue<? extends Object>... headerValues)	Отправка сообщения
ResultConsumer<Subscribe,SubscribeResult>	subscribe(Subscribe subscribe)	Подписка на события сервера Используйте возможности клиента TcpClient.subscribe(String, Consumer)
BiConsumer<ErrMessage,TcpHeader>	unbindedError()	Возвращает обработчик общих ошибок, см getErrorConsumers()
TcpProtocol	unbindedError(BiConsumer<ErrMessage,TcpHeader> cons)	Указывает обработчик общих ошибок, см getErrorConsumers()
BiConsumer<Message,TcpHeader>	unbindedMessage()	Возвращает обработчик сообщений, см getResponseConsumers()
TcpProtocol	unbindedMessage(BiConsumer<Message,TcpHeader> cons)	Указывает обработчик сообщений, см getResponseConsumers()
ResultConsumer<UnSubscribe,UnSubscribeResult>	unsubscribe(UnSubscribe subscribe)	Отписка от событий сервера Используйте возможности клиента TcpClient.unsubscribe(String), TcpClient.unsubscribe(Consumer)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

sentRawDataConsumer

public final Consumer<TcpProtocol.SentRawData> sentRawDataConsumer

socket

protected final Socket socket

Сокет с которым производится работа

getOutput

protected final Supplier<OutputStream> getOutput

Канал/поток исходящих сообщений

getInput

protected final Supplier<InputStream> getInput

Канал/поток входящих сообщений

getLogger

protected final Supplier<xyz.cofe.trambda.log.api.Logger> getLogger

Логирование

sid

protected final AtomicInteger sid

Счетчик/генератор идентификаторов сообщений

sendMessage

protected final ThreadLocal<Message> sendMessage

serverEventListeners

protected final Map<String,Set<Consumer<ServerEvent>>> serverEventListeners

Подписчики на серверные события

Constructor Detail

TcpProtocol

public TcpProtocol(Socket socket)

Конструктор

Parameters:

socket - сокет

TcpProtocol

public TcpProtocol(Socket socket,
                   Consumer<TcpProtocol.SentRawData> sentRawDataConsumer)

Конструктор

Parameters:

socket - сокет

sentRawDataConsumer - подписчик на отправленные сообщения

TcpProtocol

public TcpProtocol(Socket socket,
                   xyz.cofe.trambda.log.api.Logger logger)

Конструктор

Parameters:

socket - сокет

logger - логгер

TcpProtocol

public TcpProtocol(OutputStream outputStream,
                   InputStream inputStream,
                   xyz.cofe.trambda.log.api.Logger logger)

Конструктор

Parameters:

outputStream - исходящий поток данных

inputStream - входящий поток данных

logger - логгер

TcpProtocol

public TcpProtocol(OutputStream outputStream,
                   InputStream inputStream,
                   xyz.cofe.trambda.log.api.Logger logger,
                   Consumer<TcpProtocol.SentRawData> sentRawDataConsumer)

Конструктор

Parameters:

outputStream - исходящий поток данных

inputStream - входящий поток данных

logger - логгер

sentRawDataConsumer - подписчик на отправленные сообщения

Method Detail

sendRaw

@SafeVarargs
public final int sendRaw(String method,
                         byte[] payload,
                         Consumer<Integer> sendId,
                         HeaderValue<? extends Object>... headerValues)
                  throws IOException

Отправка сообщения

Parameters:

method - метод

payload - полезная нагрузка

sendId - идентификатор сообщения

headerValues - дополнительные заголовки сообщения

Returns:

идентификатор сообщения, см TcpHeader.getSid()

Throws:

IOException - ошибка сети

send

@SafeVarargs
public final int send(Message message,
                      HeaderValue<? extends Object>... headerValues)
               throws IOException

Отправка сообщения

Parameters:

message - сообщение

headerValues - дополнительные заголовки сообщения

Returns:

идентификатор сообщения, см TcpHeader.getSid()

Throws:

IOException - ошибка сети

send

@SafeVarargs
public final int send(Message message,
                      Consumer<Integer> sid,
                      HeaderValue<? extends Object>... headerValues)
               throws IOException

Отправка сообщения

Parameters:

message - сообщение

sid - идентификатор сообщения

headerValues - дополнительные заголовки сообщения

Returns:

идентификатор сообщения, см TcpHeader.getSid()

Throws:

IOException - ошибка сети

receiveRaw

public Optional<xyz.cofe.fn.Tuple2<RawPack,Optional<BuffInputStream>>> receiveRaw(Optional<BuffInputStream> prevStream)
                                                                                 throws IOException

Получение сообщения из сети

Parameters:

prevStream - предыдущая часть сообщения

Returns:

сообщение или входящий поток закрыт

Throws:

IOException - ошибка сети

getErrorConsumers

public Map<Integer,Consumer<ErrMessage>> getErrorConsumers()

Возвращает обработчики ошибок на конкретные запросы TcpHeader.getSid()

Returns:

Карта ключ - идентификатор сообщения TcpHeader.getSid() / обработчик

getResponseConsumers

public Map<Integer,Consumer<? extends Message>> getResponseConsumers()

Возвращает обработчики на конкретные запросы TcpHeader.getSid()

Returns:

Карта ключ - идентификатор сообщения TcpHeader.getSid() / обработчик

unbindedError

public BiConsumer<ErrMessage,TcpHeader> unbindedError()

Возвращает обработчик общих ошибок, см getErrorConsumers()

Returns:

обработчик общих ошибок

unbindedError

public TcpProtocol unbindedError(BiConsumer<ErrMessage,TcpHeader> cons)

Указывает обработчик общих ошибок, см getErrorConsumers()

Parameters:

cons - обработчик общих ошибок

Returns:

SELF ссылка

unbindedMessage

public BiConsumer<Message,TcpHeader> unbindedMessage()

Возвращает обработчик сообщений, см getResponseConsumers()

Returns:

обработчик сообщений

unbindedMessage

public TcpProtocol unbindedMessage(BiConsumer<Message,TcpHeader> cons)

Указывает обработчик сообщений, см getResponseConsumers()

Parameters:

cons - обработчик сообщений

Returns:

SELF ссылка

readNow

public boolean readNow()
                throws IOException

Читает сообщение Message из входного потока (getInput).

• Делает попытку чтения данных из входящего потока receiveRaw(Optional)
• Проверяет контрольную сумму сообщения RawPackReadonly.isPayloadChecksumMatched()
• Вычленяет из пакета байтов само сообщение RawPackReadonly.payloadMessage()
• Отправляет сообщение на последующую обработку process(Message, TcpHeader)

Returns:

true - сообщение прочитано

Throws:

IOException - Ошибка чтения данных

process

protected void process(Message msg,
                       TcpHeader header)

Обработка входящего сообщения.

• Отрабатывает сообщения Ping - вызывает process(Ping, TcpHeader)
• Отрабатывает сообщения Pong - вызывает process(Pong, TcpHeader)
• Отрабатывает сообщения ErrMessage
  • Если указан на какое сообщение TcpHeader.getReferrer() ответ
    • то вызывает обработчик конкретный getErrorConsumers()
    • или общий обработчик unbindedError()
• Обрабатывает сообщение ServerEvent и вызывает process(ServerEvent, TcpHeader)
• Для всех остальных сообщений
  • Если есть зарегистрированный обработчик getResponseConsumers(), то вызывается он
  • Либо общий обработчик unbindedMessage()

Parameters:

msg - входящие сообщение

header - заголовки сообщения

process

protected void process(Pong pong,
                       TcpHeader header)

При получении ответа Pong уведомляет подписанотов ping(Consumer), тех кто отправил запрос Ping

Parameters:

pong - Ответ

header - заголовки

process

protected void process(Ping ping,
                       TcpHeader header)

При получении Ping сообщения, отсылает ответ Pong

Parameters:

ping - Входящий Ping

header - Заголовки

listenServerEvent

public AutoCloseable listenServerEvent(String publisher,
                                       Consumer<ServerEvent> listener)

Добавляет подписчика на событие сервера.
Используйте возможности клиента TcpClient.subscribe(String, Consumer)

Parameters:

publisher - издатель события

listener - подписчик

Returns:

отписка от событий

removeServerEventListener

public void removeServerEventListener(Consumer<ServerEvent> listener)

Отписка от событий сервера

Parameters:

listener - подписчик

process

protected void process(ServerEvent sevent,
                       TcpHeader header)

Обработка входящего серверного события

• Уведомляет подписчиков listenServerEvent(String, Consumer) о серверном событии

Parameters:

sevent - серверное событие

header - заголовки

ping

public int ping(Consumer<Pong> consumer)

Отправляет Ping запрос

Parameters:

consumer - приемник ответа

Returns:

Идентификатор TcpHeader.getSid()

compile

public ResultConsumer<Compile,CompileResult> compile(xyz.cofe.trambda.LambdaDump methodDef)

Компиляция лямбды

Parameters:

methodDef - лямбда

Returns:

Отправка запроса

execute

public ResultConsumer<Execute,ExecuteResult> execute(CompileResult cres)

Выполнение ранее скомпилированной лямбды

Parameters:

cres - ранее скомпилированная лямбда compile(LambdaDump)

Returns:

Результат выполнения

subscribe

public ResultConsumer<Subscribe,SubscribeResult> subscribe(Subscribe subscribe)

Подписка на события сервера
Используйте возможности клиента TcpClient.subscribe(String, Consumer)

Parameters:

subscribe - имя издателя сервера

Returns:

Результат подписки

unsubscribe

public ResultConsumer<UnSubscribe,UnSubscribeResult> unsubscribe(UnSubscribe subscribe)

Отписка от событий сервера
Используйте возможности клиента TcpClient.unsubscribe(String), TcpClient.unsubscribe(Consumer)

Parameters:

subscribe - подписчик

Returns:

Результат отписки