datablockprotocol.h

/*******************************************************************************
 * Copyright (c) 2018 Nerian Vision GmbH
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *******************************************************************************/

#ifndef VISIONTRANSFER_DATABLOCKPROTOCOL_H
#define VISIONTRANSFER_DATABLOCKPROTOCOL_H

#include <map>
#include <vector>
#include <memory>
#include <chrono>
#include <deque>

#include "visiontransfer/alignedallocator.h"

class DataBlockProtocol {
public:
    enum ProtocolType {
        PROTOCOL_TCP,
        PROTOCOL_UDP
    };

    // Constants that are also used in other places.
    static const int MAX_TCP_BYTES_TRANSFER = 0xFFFF; //64K - 1
    static const int MAX_UDP_RECEPTION = 0x4000; //16K
    static const int MAX_OUTSTANDING_BYTES = 2*MAX_TCP_BYTES_TRANSFER;

    DataBlockProtocol(bool server, ProtocolType protType, int maxUdpPacketSize);

    int getProtocolOverhead() const {
        return protType == PROTOCOL_UDP ? sizeof(int) : 0;
    }

    int getMaxReceptionSize() const;

    void resetTransfer();

    void setTransferHeader(unsigned char* data, int headerSize, int transferSize);

    void setTransferData(unsigned char* data,  int validBytes = 0x7FFFFFFF);

    void setTransferValidBytes(int validBytes);

    const unsigned char* getTransferMessage(int& length);

    bool transferComplete();

    unsigned char* getNextReceiveBuffer(int maxLength);

    void resetReception(bool dropped);

    void processReceivedMessage(int length, bool& transferComplete);

    unsigned char* getReceivedData(int& length);

    unsigned char* getReceivedHeader(int& length);

    int getDroppedReceptions() const {
        return droppedReceptions;
    }

    bool newClientConnected();

    bool isConnected() const;

    const unsigned char* getNextControlMessage(int& length);

private:
    // The pimpl idiom is not necessary here, as this class is usually not
    // used directly

    struct MissingReceiveSegment {
        int offset;
        int length;
        bool isEof;
        unsigned char subsequentData[4];
    };

    static constexpr int HEARTBEAT_INTERVAL_MS = 1000;
    static constexpr int RECONNECT_TIMEOUT_MS = 1000;

    static constexpr unsigned char CONNECTION_MESSAGE = 0x01;
    static constexpr unsigned char CONFIRM_MESSAGE = 0x02;
    static constexpr unsigned char HEADER_MESSAGE = 0x03;
    static constexpr unsigned char RESEND_MESSAGE = 0x04;
    static constexpr unsigned char EOF_MESSAGE = 0x05;
    static constexpr unsigned char HEARTBEAT_MESSAGE = 0x06;

    bool isServer;
    ProtocolType protType;
    int maxPayloadSize;
    int minPayloadSize;

    // Transfer related variables
    bool transferDone;
    unsigned char* rawData;
    int rawValidBytes;
    int transferOffset;
    int transferSize;
    int overwrittenTransferData;
    int overwrittenTransferIndex;
    unsigned char* transferHeaderData;
    int transferHeaderSize;

    // Reliability related variables
    std::deque<MissingReceiveSegment> missingReceiveSegments;
    std::deque<std::pair<int, int> > missingTransferSegments;
    bool waitingForMissingSegments;
    int totalReceiveSize;

    unsigned char controlMessageBuffer[1024];

    // Connection related variables
    bool connectionConfirmed;
    bool confirmationMessagePending;
    bool eofMessagePending;
    bool clientConnectionPending;
    bool resendMessagePending;
    std::chrono::steady_clock::time_point lastRemoteHostActivity;
    std::chrono::steady_clock::time_point lastSentHeartbeat;
    std::chrono::steady_clock::time_point lastReceivedHeartbeat;

    // Reception related variables
    std::vector<unsigned char, AlignedAllocator<unsigned char> > receiveBuffer;
    int receiveDataSize;
    int receiveOffset;
    std::vector<unsigned char> receivedHeader;
    bool finishedReception;
    int droppedReceptions;
    unsigned char unprocessedMsgPart[MAX_OUTSTANDING_BYTES];
    int unprocessedMsgLength;
    bool headerReceived;

    const unsigned char* extractPayload(const unsigned char* data, int& length, bool& error);
    bool processControlMessage(int length);
    void restoreTransferBuffer();
    bool generateResendRequest(int& length);
    void getNextTransferSegment(int& offset, int& length);
    void parseResendMessage(int length);
    void parseEofMessage(int length);
    int getNextUdpReceiveOffset(int lastSegmentOffset, int lastSegmentSize);
    void processReceivedUdpMessage(int length, bool& transferComplete);
    void processReceivedTcpMessage(int length, bool& transferComplete);
    void resizeReceiveBuffer();
    int parseReceivedHeader(int length, int offset);
};

#endif