pip/libs/main/io_utils/pibasetransfer.h

/*! \file pibasetransfer.h
 * \ingroup IO-Utils
 * \~\brief
 * \~english Base class for reliable packet sessions with acknowledgements, pause and resume
 * \~russian Базовый класс для надежных пакетных сессий с подтверждениями, паузой и продолжением
 */
/*
    PIP - Platform Independent Primitives
    Base class for reliable send and receive data in fixed packets with error correction, pause and resume
    Andrey Bychkov work.a.b@yandex.ru

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#ifndef PIBASETRANSFER_H
#define PIBASETRANSFER_H

#include "picrc.h"
#include "pidiagnostics.h"

//! \ingroup IO-Utils
//! \~\brief
//! \~english Base transport for reliable fixed-size packet exchange over an external channel.
//! \~russian Базовый транспорт для надежного обмена пакетами фиксированного размера через внешний канал.
class PIP_EXPORT PIBaseTransfer: public PIObject {
	PIOBJECT_SUBCLASS(PIBaseTransfer, PIObject);


public:
	//! \~english Constructs transfer with default packet size, timeout and diagnostics.
	//! \~russian Создает передачу с размером пакета, таймаутом и диагностикой по умолчанию.
	PIBaseTransfer();

	//! \~english Stops active transfer state and owned diagnostics.
	//! \~russian Останавливает активное состояние передачи и встроенную диагностику.
	~PIBaseTransfer();


#pragma pack(push, 1)
	//! \~english Common header placed before every protocol packet.
	//! \~russian Общий заголовок, размещаемый перед каждым пакетом протокола.
	struct PIP_EXPORT PacketHeader {
		//! \~english Packet signature used to recognize transfer packets.
		//! \~russian Сигнатура пакета, используемая для распознавания пакетов передачи.
		uint sig;
		//! \~english Packet type from the transfer protocol.
		//! \~russian Тип пакета из протокола передачи.
		int type; // PacketType
		//! \~english Session identifier for current transfer exchange.
		//! \~russian Идентификатор сессии текущего обмена.
		int session_id;
		//! \~english Sequential packet identifier inside the session.
		//! \~russian Последовательный идентификатор пакета внутри сессии.
		uint id;
		//! \~english CRC calculated for the packet payload.
		//! \~russian CRC, вычисленная для полезной нагрузки пакета.
		uint crc;
		//! \~english Returns whether the packet signature matches \a packetSignature().
		//! \~russian Возвращает, совпадает ли сигнатура пакета с \a packetSignature().
		bool check_sig() { return (sig == signature); }
	};

	//! \~english Logical fragment description used to split data into packets.
	//! \~russian Описание логического фрагмента, используемое для разбиения данных на пакеты.
	struct PIP_EXPORT Part {
		//! \~english Constructs part metadata for item "id_", fragment size "size_" and offset "start_".
		//! \~russian Создает метаданные части для элемента "id_", размера фрагмента "size_" и смещения "start_".
		Part(uint id_ = 0, ullong size_ = 0, ullong start_ = 0): id(id_), size(size_), start(start_) {}
		//! \~english Identifier of the logical item being transferred.
		//! \~russian Идентификатор логического элемента, который передается.
		uint id;
		//! \~english Size of this fragment in bytes.
		//! \~russian Размер этого фрагмента в байтах.
		ullong size;
		//! \~english Byte offset of this fragment inside the logical item.
		//! \~russian Смещение этого фрагмента в байтах внутри логического элемента.
		ullong start;
	};
#pragma pack(pop)


	//! \~english Requests cancellation of the current send session.
	//! \~russian Запрашивает отмену текущей сессии отправки.
	void stopSend();

	//! \~english Requests cancellation of the current receive session.
	//! \~russian Запрашивает отмену текущей сессии приема.
	void stopReceive();


	//! \~english Returns whether a send session is active.
	//! \~russian Возвращает, активна ли сессия отправки.
	bool isSending() const { return is_sending; }

	//! \~english Returns whether a receive session is active.
	//! \~russian Возвращает, активна ли сессия приема.
	bool isReceiving() const { return is_receiving; }

	//! \~english Returns whether the transfer is currently paused.
	//! \~russian Возвращает, находится ли передача сейчас на паузе.
	bool isPause() const { return is_pause; }

	//! \~english Switches current session between paused and resumed states.
	//! \~russian Переключает текущую сессию между состояниями паузы и продолжения.
	void setPause(bool pause_);


	//! \~english Sets maximum encoded packet size.
	//! \~russian Устанавливает максимальный размер кодированного пакета.
	void setPacketSize(int size) { packet_size = size; }
	//! \~english Returns maximum encoded packet size.
	//! \~russian Возвращает максимальный размер кодированного пакета.
	int packetSize() const { return packet_size; }

	//! \~english Sets session timeout for start negotiation, acknowledgements and pause recovery.
	//! \~russian Устанавливает таймаут сессии для согласования старта, подтверждений и восстановления после паузы.
	void setTimeout(double sec);
	//! \~english Returns session timeout in seconds.
	//! \~russian Возвращает таймаут сессии в секундах.
	double timeout() const { return timeout_; }

	//! \~english Enables or disables CRC validation for data packets.
	//! \~russian Включает или выключает проверку CRC для пакетов данных.
	void setCRCEnabled(bool en = true) { crc_enabled = en; }
	//! \~english Returns whether CRC validation is enabled.
	//! \~russian Возвращает, включена ли проверка CRC.
	bool isCRCEnabled() const { return crc_enabled; }

	//! \~english Returns short textual state of the current session.
	//! \~russian Возвращает краткое текстовое состояние текущей сессии.
	PIString stateString() const { return state_string; }
	//! \~english Returns acknowledgement map for the current session.
	//! \~russian Возвращает карту подтверждений для текущей сессии.
	PIString packetMap() const { return pm_string; }
	//! \~english Returns total number of bytes planned for the current session.
	//! \~russian Возвращает общее число байтов, запланированных для текущей сессии.
	llong bytesAll() const { return bytes_all; }
	//! \~english Returns number of bytes already processed in the current session.
	//! \~russian Возвращает число байтов, уже обработанных в текущей сессии.
	llong bytesCur() const { return bytes_cur; }
	//! \~english Returns live transport diagnostics.
	//! \~russian Возвращает текущую диагностику транспорта.
	const PIDiagnostics & diagnostic() { return diag; }

	//! \~english Returns the packet signature constant used by the protocol.
	//! \~russian Возвращает константу сигнатуры пакета, используемую протоколом.
	static uint packetSignature() { return signature; }

	EVENT_HANDLER1(void, received, PIByteArray, data);
	EVENT_HANDLER(void, stop) {
		stopSend();
		stopReceive();
	}
	EVENT_HANDLER(void, pause) { setPause(true); }
	EVENT_HANDLER(void, resume) { setPause(false); }

	EVENT(receiveStarted);
	EVENT(paused);
	EVENT(resumed);
	EVENT1(receiveFinished, bool, ok);
	EVENT(sendStarted);
	EVENT1(sendFinished, bool, ok);
	EVENT1(sendRequest, PIByteArray &, data);

	//! \handlers
	//! \{
	//!
	//! \fn void received(PIByteArray data)
	//! \~english Feeds one encoded packet received from the external transport.
	//! \~russian Передает один закодированный пакет, полученный от внешнего транспорта.
	//!
	//! \fn void stop()
	//! \~english Stops both sending and receiving sides of the current session.
	//! \~russian Останавливает и отправку, и прием текущей сессии.
	//!
	//! \fn void pause()
	//! \~english Switches current session to paused state.
	//! \~russian Переводит текущую сессию в состояние паузы.
	//!
	//! \fn void resume()
	//! \~english Resumes the current paused session.
	//! \~russian Продолжает текущую приостановленную сессию.
	//!
	//! \}
	//! \events
	//! \{
	//!
	//! \fn void receiveStarted()
	//! \~english Emitted when a receive session is accepted and initialized.
	//! \~russian Генерируется, когда сессия приема принята и инициализирована.
	//!
	//! \fn void paused()
	//! \~english Emitted when the transfer enters paused state.
	//! \~russian Генерируется, когда передача переходит в состояние паузы.
	//!
	//! \fn void resumed()
	//! \~english Emitted when the transfer leaves paused state.
	//! \~russian Генерируется, когда передача выходит из состояния паузы.
	//!
	//! \fn void receiveFinished(bool ok)
	//! \~english Emitted when the receive session finishes with result "ok".
	//! \~russian Генерируется, когда сессия приема завершается с результатом "ok".
	//!
	//! \fn void sendStarted()
	//! \~english Emitted when a prepared send session starts.
	//! \~russian Генерируется при запуске подготовленной сессии отправки.
	//!
	//! \fn void sendFinished(bool ok)
	//! \~english Emitted when the send session finishes with result "ok".
	//! \~russian Генерируется, когда сессия отправки завершается с результатом "ok".
	//!
	//! \fn void sendRequest(PIByteArray data)
	//! \~english Emitted for every encoded packet that must be written to the external transport.
	//! \~russian Генерируется для каждого закодированного пакета, который нужно записать во внешний транспорт.
	//!
	//! \}

protected:
	//! \~english Builds session packet layout for logical parts in "parts".
	//! \~russian Формирует раскладку пакетов сессии для логических частей из "parts".
	void buildSession(PIVector<Part> parts);
	//! \~english Returns payload bytes for one requested logical fragment.
	//! \~russian Возвращает байты полезной нагрузки для одного запрошенного логического фрагмента.
	virtual PIByteArray buildPacket(Part fi)                               = 0;
	//! \~english Consumes one received logical fragment and optional custom packet header.
	//! \~russian Обрабатывает один принятый логический фрагмент и необязательный пользовательский заголовок пакета.
	virtual void receivePart(Part fi, PIByteArray ba, PIByteArray pheader) = 0;
	//! \~english Called after a new receive session is accepted and initialized.
	//! \~russian Вызывается после принятия и инициализации новой сессии приема.
	virtual void beginReceive() { ; }
	//! \~english Returns custom header bytes inserted after \a PacketHeader in each data packet.
	//! \~russian Возвращает байты пользовательского заголовка, вставляемые после \a PacketHeader в каждом пакете данных.
	virtual PIByteArray customHeader() { return PIByteArray(); }
	//! \~english Runs the prepared send session until success, failure or cancellation.
	//! \~russian Выполняет подготовленную сессию отправки до успеха, ошибки или отмены.
	bool send_process();

	uint packet_header_size, part_header_size;
	bool break_, is_sending, is_receiving, is_pause;
	PIString state_string;
	llong bytes_all, bytes_cur;

private:
	enum PacketType {
		pt_Unknown,
		pt_Data,
		pt_ReplySuccess,
		pt_ReplyInvalid,
		pt_Break,
		pt_Start,
		pt_Pause
	};

#pragma pack(push, 1)
	struct PIP_EXPORT StartRequest {
		uint packets;
		ullong size;
	};
#pragma pack(pop)

	BINARY_STREAM_FRIEND(PIBaseTransfer::StartRequest);

	void processData(int id, PIByteArray & data);
	PIByteArray build_packet(int id);
	int checkSession();
	void sendBreak(int session_id);
	void sendReply(PacketType reply);
	bool getStartRequest();
	bool finish_send(bool ok);
	void finish_receive(bool ok, bool quet = false);

	static const uint signature;
	int packet_size, packets_count;
	double timeout_;
	PIVector<PIVector<Part>> session;
	PIVector<PacketType> replies;
	PITimeMeasurer send_tm, pause_tm;
	PacketHeader header;
	CRC_16 crc;
	int send_queue;
	int send_up;
	PIDiagnostics diag;
	PIMutex mutex_session;
	PIMutex mutex_send;
	PIMutex mutex_header;
	bool crc_enabled;
	PIString pm_string;
};


BINARY_STREAM_WRITE(PIBaseTransfer::PacketHeader) {
	s << v.sig << v.type << v.session_id << v.id << v.crc;
	return s;
}
BINARY_STREAM_READ(PIBaseTransfer::PacketHeader) {
	s >> v.sig >> v.type >> v.session_id >> v.id >> v.crc;
	return s;
}

BINARY_STREAM_WRITE(PIBaseTransfer::Part) {
	s << v.id << v.size << v.start;
	return s;
}
BINARY_STREAM_READ(PIBaseTransfer::Part) {
	s >> v.id >> v.size >> v.start;
	return s;
}

BINARY_STREAM_WRITE(PIBaseTransfer::StartRequest) {
	s << v.packets << v.size;
	return s;
}
BINARY_STREAM_READ(PIBaseTransfer::StartRequest) {
	s >> v.packets >> v.size;
	return s;
}

inline PICout operator<<(PICout s, const PIBaseTransfer::Part & v) {
	s.saveAndSetControls(0);
	s << "Part(\"" << v.id << "\", " << PIString::readableSize(v.start) << " b | " << PIString::readableSize(v.size) << " b)";
	s.restoreControls();
	return s;
}


#endif // PIBASETRANSFER_H