/*
* Copyright (c) 2013-2022, The PurpleI2P Project
*
* This file is part of Purple i2pd project and licensed under BSD3
*
* See full license text in LICENSE file at top of project tree
*/

#include "Crypto.h"
#include "Log.h"
#include "RouterInfo.h"
#include "RouterContext.h"
#include "Tunnel.h"
#include "Timestamp.h"
#include "Destination.h"
#include "Streaming.h"

namespace i2p {
    namespace stream {
        void SendBufferQueue::Add(const uint8_t *buf, size_t len, SendHandler handler) {
            Add(std::make_shared<SendBuffer>(buf, len, handler));
        }

        void SendBufferQueue::Add(std::shared_ptr<SendBuffer> buf) {
            if (buf) {
                m_Buffers.push_back(buf);
                m_Size += buf->len;
            }
        }

        size_t SendBufferQueue::Get(uint8_t *buf, size_t len) {
            size_t offset = 0;
            while (!m_Buffers.empty() && offset < len) {
                auto nextBuffer = m_Buffers.front();
                auto rem = nextBuffer->GetRemainingSize();
                if (offset + rem <= len) {
                    // whole buffer
                    memcpy(buf + offset, nextBuffer->GetRemaningBuffer(), rem);
                    offset += rem;
                    m_Buffers.pop_front(); // delete it
                } else {
                    // partially
                    rem = len - offset;
                    memcpy(buf + offset, nextBuffer->GetRemaningBuffer(), len - offset);
                    nextBuffer->offset += (len - offset);
                    offset = len; // break
                }
            }
            m_Size -= offset;
            return offset;
        }

        void SendBufferQueue::CleanUp() {
            if (!m_Buffers.empty()) {
                for (auto it: m_Buffers)
                    it->Cancel();
                m_Buffers.clear();
                m_Size = 0;
            }
        }

        Stream::Stream(boost::asio::io_service &service, StreamingDestination &local,
                       std::shared_ptr<const i2p::data::LeaseSet> remote, int port) : m_Service(service),
                                                                                      m_SendStreamID(0),
                                                                                      m_SequenceNumber(0),
                                                                                      m_LastReceivedSequenceNumber(-1),
                                                                                      m_Status(eStreamStatusNew),
                                                                                      m_IsAckSendScheduled(false),
                                                                                      m_LocalDestination(local),
                                                                                      m_RemoteLeaseSet(remote),
                                                                                      m_ReceiveTimer(m_Service),
                                                                                      m_ResendTimer(m_Service),
                                                                                      m_AckSendTimer(m_Service),
                                                                                      m_NumSentBytes(0),
                                                                                      m_NumReceivedBytes(0),
                                                                                      m_Port(port),
                                                                                      m_WindowSize(MIN_WINDOW_SIZE),
                                                                                      m_RTT(INITIAL_RTT),
                                                                                      m_RTO(INITIAL_RTO),
                                                                                      m_AckDelay(
                                                                                              local.GetOwner()->GetStreamingAckDelay()),
                                                                                      m_LastWindowSizeIncreaseTime(0),
                                                                                      m_NumResendAttempts(0),
                                                                                      m_MTU(STREAMING_MTU) {
            RAND_bytes((uint8_t * ) & m_RecvStreamID, 4);
            m_RemoteIdentity = remote->GetIdentity();
        }

        Stream::Stream(boost::asio::io_service &service, StreamingDestination &local) :
                m_Service(service), m_SendStreamID(0), m_SequenceNumber(0), m_LastReceivedSequenceNumber(-1),
                m_Status(eStreamStatusNew), m_IsAckSendScheduled(false), m_LocalDestination(local),
                m_ReceiveTimer(m_Service), m_ResendTimer(m_Service), m_AckSendTimer(m_Service),
                m_NumSentBytes(0), m_NumReceivedBytes(0), m_Port(0), m_WindowSize(MIN_WINDOW_SIZE),
                m_RTT(INITIAL_RTT), m_RTO(INITIAL_RTO), m_AckDelay(local.GetOwner()->GetStreamingAckDelay()),
                m_LastWindowSizeIncreaseTime(0), m_NumResendAttempts(0), m_MTU(STREAMING_MTU) {
            RAND_bytes((uint8_t * ) & m_RecvStreamID, 4);
        }

        Stream::~Stream() {
            CleanUp();
            LogPrint(eLogDebug, "Streaming: Stream deleted");
        }

        void Stream::Terminate(bool deleteFromDestination) // should be called from StreamingDestination::Stop only
        {
            m_Status = eStreamStatusTerminated;
            m_AckSendTimer.cancel();
            m_ReceiveTimer.cancel();
            m_ResendTimer.cancel();
            //CleanUp (); /* Need to recheck - broke working on windows */
            if (deleteFromDestination)
                m_LocalDestination.DeleteStream(shared_from_this());
        }

        void Stream::CleanUp() {
            {
                std::unique_lock<std::mutex> l(m_SendBufferMutex);
                m_SendBuffer.CleanUp();
            }
            while (!m_ReceiveQueue.empty()) {
                auto packet = m_ReceiveQueue.front();
                m_ReceiveQueue.pop();
                m_LocalDestination.DeletePacket(packet);
            }

            for (auto it: m_SentPackets)
                m_LocalDestination.DeletePacket(it);
            m_SentPackets.clear();

            for (auto it: m_SavedPackets)
                m_LocalDestination.DeletePacket(it);
            m_SavedPackets.clear();
        }

        void Stream::HandleNextPacket(Packet *packet) {
            m_NumReceivedBytes += packet->GetLength();
            if (!m_SendStreamID)
                m_SendStreamID = packet->GetReceiveStreamID();

            if (!packet->IsNoAck()) // ack received
                ProcessAck(packet);

            int32_t receivedSeqn = packet->GetSeqn();
            bool isSyn = packet->IsSYN();
            if (!receivedSeqn && !isSyn) {
                // plain ack
                LogPrint(eLogDebug, "Streaming: Plain ACK received");
                m_LocalDestination.DeletePacket(packet);
                return;
            }

            LogPrint(eLogDebug, "Streaming: Received seqn=", receivedSeqn, " on sSID=", m_SendStreamID);
            if (receivedSeqn == m_LastReceivedSequenceNumber + 1) {
                // we have received next in sequence message
                ProcessPacket(packet);

                // we should also try stored messages if any
                for (auto it = m_SavedPackets.begin(); it != m_SavedPackets.end();) {
                    if ((*it)->GetSeqn() == (uint32_t)(m_LastReceivedSequenceNumber + 1)) {
                        Packet *savedPacket = *it;
                        m_SavedPackets.erase(it++);

                        ProcessPacket(savedPacket);
                    } else
                        break;
                }

                // schedule ack for last message
                if (m_Status == eStreamStatusOpen) {
                    if (!m_IsAckSendScheduled) {
                        m_IsAckSendScheduled = true;
                        auto ackTimeout = m_RTT / 10;
                        if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay;
                        else if (ackTimeout < MIN_SEND_ACK_TIMEOUT) ackTimeout = MIN_SEND_ACK_TIMEOUT;
                        m_AckSendTimer.expires_from_now(boost::posix_time::milliseconds(ackTimeout));
                        m_AckSendTimer.async_wait(std::bind(&Stream::HandleAckSendTimer,
                                                            shared_from_this(), std::placeholders::_1));
                    }
                } else if (isSyn)
                    // we have to send SYN back to incoming connection
                    SendBuffer(); // also sets m_IsOpen
            } else {
                if (receivedSeqn <= m_LastReceivedSequenceNumber) {
                    // we have received duplicate
                    LogPrint(eLogWarning, "Streaming: Duplicate message ", receivedSeqn, " on sSID=", m_SendStreamID);
                    SendQuickAck(); // resend ack for previous message again
                    m_LocalDestination.DeletePacket(packet); // packet dropped
                } else {
                    LogPrint(eLogWarning, "Streaming: Missing messages on sSID=", m_SendStreamID, ": from ",
                             m_LastReceivedSequenceNumber + 1, " to ", receivedSeqn - 1);
                    // save message and wait for missing message again
                    SavePacket(packet);
                    if (m_LastReceivedSequenceNumber >= 0) {
                        // send NACKs for missing messages ASAP
                        if (m_IsAckSendScheduled) {
                            m_IsAckSendScheduled = false;
                            m_AckSendTimer.cancel();
                        }
                        SendQuickAck();
                    } else {
                        // wait for SYN
                        m_IsAckSendScheduled = true;
                        m_AckSendTimer.expires_from_now(boost::posix_time::milliseconds(SYN_TIMEOUT));
                        m_AckSendTimer.async_wait(std::bind(&Stream::HandleAckSendTimer,
                                                            shared_from_this(), std::placeholders::_1));
                    }
                }
            }
        }

        void Stream::SavePacket(Packet *packet) {
            if (!m_SavedPackets.insert(packet).second)
                m_LocalDestination.DeletePacket(packet);
        }

        void Stream::ProcessPacket(Packet *packet) {
            uint32_t receivedSeqn = packet->GetSeqn();
            uint16_t flags = packet->GetFlags();
            LogPrint(eLogDebug, "Streaming: Process seqn=", receivedSeqn, ", flags=", flags);

            if (!ProcessOptions(flags, packet)) {
                m_LocalDestination.DeletePacket(packet);
                Terminate();
                return;
            }

            packet->offset = packet->GetPayload() - packet->buf;
            if (packet->GetLength() > 0) {
                m_ReceiveQueue.push(packet);
                m_ReceiveTimer.cancel();
            } else
                m_LocalDestination.DeletePacket(packet);

            m_LastReceivedSequenceNumber = receivedSeqn;

            if (flags & PACKET_FLAG_RESET) {
                LogPrint(eLogDebug, "Streaming: closing stream sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID,
                         ": reset flag received in packet #", receivedSeqn);
                m_Status = eStreamStatusReset;
                Close();
            } else if (flags & PACKET_FLAG_CLOSE) {
                if (m_Status != eStreamStatusClosed)
                    SendClose();
                m_Status = eStreamStatusClosed;
                Terminate();
            }
        }

        bool Stream::ProcessOptions(uint16_t flags, Packet *packet) {
            const uint8_t *optionData = packet->GetOptionData();
            size_t optionSize = packet->GetOptionSize();
            if (flags & PACKET_FLAG_DELAY_REQUESTED) {
                if (!m_IsAckSendScheduled) {
                    uint16_t delayRequested = bufbe16toh(optionData);
                    if (delayRequested > 0 && delayRequested < m_RTT) {
                        m_IsAckSendScheduled = true;
                        m_AckSendTimer.expires_from_now(boost::posix_time::milliseconds(delayRequested));
                        m_AckSendTimer.async_wait(std::bind(&Stream::HandleAckSendTimer,
                                                            shared_from_this(), std::placeholders::_1));
                    }
                }
                optionData += 2;
            }

            if (flags & PACKET_FLAG_FROM_INCLUDED) {
                if (m_RemoteLeaseSet) m_RemoteIdentity = m_RemoteLeaseSet->GetIdentity();
                if (!m_RemoteIdentity)
                    m_RemoteIdentity = std::make_shared<i2p::data::IdentityEx>(optionData, optionSize);
                if (m_RemoteIdentity->IsRSA()) {
                    LogPrint(eLogInfo, "Streaming: Incoming stream from RSA destination ",
                             m_RemoteIdentity->GetIdentHash().ToBase64(), " Discarded");
                    return false;
                }
                optionData += m_RemoteIdentity->GetFullLen();
                if (!m_RemoteLeaseSet)
                    LogPrint(eLogDebug, "Streaming: Incoming stream from ", m_RemoteIdentity->GetIdentHash().ToBase64(),
                             ", sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID);
            }

            if (flags & PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED) {
                uint16_t maxPacketSize = bufbe16toh(optionData);
                LogPrint(eLogDebug, "Streaming: Max packet size ", maxPacketSize);
                optionData += 2;
            }

            if (flags & PACKET_FLAG_OFFLINE_SIGNATURE) {
                if (!m_RemoteIdentity) {
                    LogPrint(eLogInfo, "Streaming: offline signature without identity");
                    return false;
                }
                // if we have it in LeaseSet already we don't need to parse it again
                if (m_RemoteLeaseSet) m_TransientVerifier = m_RemoteLeaseSet->GetTransientVerifier();
                if (m_TransientVerifier) {
                    // skip option data
                    optionData += 6; // timestamp and key type
                    optionData += m_TransientVerifier->GetPublicKeyLen(); // public key
                    optionData += m_RemoteIdentity->GetSignatureLen(); // signature
                } else {
                    // transient key
                    size_t offset = 0;
                    m_TransientVerifier = i2p::data::ProcessOfflineSignature(m_RemoteIdentity, optionData, optionSize -
                                                                                                           (optionData -
                                                                                                            packet->GetOptionData()),
                                                                             offset);
                    optionData += offset;
                    if (!m_TransientVerifier) {
                        LogPrint(eLogError, "Streaming: offline signature failed");
                        return false;
                    }
                }
            }

            if (flags & PACKET_FLAG_SIGNATURE_INCLUDED) {
                uint8_t signature[256];
                auto signatureLen = m_TransientVerifier ? m_TransientVerifier->GetSignatureLen()
                                                        : m_RemoteIdentity->GetSignatureLen();
                if (signatureLen <= sizeof(signature)) {
                    memcpy(signature, optionData, signatureLen);
                    memset(const_cast<uint8_t *>(optionData), 0, signatureLen);
                    bool verified = m_TransientVerifier ?
                                    m_TransientVerifier->Verify(packet->GetBuffer(), packet->GetLength(), signature) :
                                    m_RemoteIdentity->Verify(packet->GetBuffer(), packet->GetLength(), signature);
                    if (!verified) {
                        LogPrint(eLogError, "Streaming: Signature verification failed, sSID=", m_SendStreamID,
                                 ", rSID=", m_RecvStreamID);
                        Close();
                        flags |= PACKET_FLAG_CLOSE;
                    }
                    memcpy(const_cast<uint8_t *>(optionData), signature, signatureLen);
                    optionData += signatureLen;
                } else {
                    LogPrint(eLogError, "Streaming: Signature too big, ", signatureLen, " bytes");
                    return false;
                }
            }
            return true;
        }

        void Stream::HandlePing(Packet *packet) {
            uint16_t flags = packet->GetFlags();
            if (ProcessOptions(flags, packet) && m_RemoteIdentity) {
                // send pong
                Packet p;
                memset(p.buf, 0, 22); // minimal header all zeroes
                memcpy(p.buf + 4, packet->buf, 4); // but receiveStreamID is the sendStreamID from the ping
                htobe16buf(p.buf + 18, PACKET_FLAG_ECHO); // and echo flag
                ssize_t payloadLen = packet->len - (packet->GetPayload() - packet->buf);
                if (payloadLen > 0)
                    memcpy(p.buf + 22, packet->GetPayload(), payloadLen);
                else
                    payloadLen = 0;
                p.len = payloadLen + 22;
                SendPackets(std::vector<Packet *>{&p});
                LogPrint(eLogDebug, "Streaming: Pong of ", p.len, " bytes sent");
            }
            m_LocalDestination.DeletePacket(packet);
        }

        void Stream::ProcessAck(Packet *packet) {
            bool acknowledged = false;
            auto ts = i2p::util::GetMillisecondsSinceEpoch();
            uint32_t ackThrough = packet->GetAckThrough();
            if (ackThrough > m_SequenceNumber) {
                LogPrint(eLogError, "Streaming: Unexpected ackThrough=", ackThrough, " > seqn=", m_SequenceNumber);
                return;
            }
            int nackCount = packet->GetNACKCount();
            for (auto it = m_SentPackets.begin(); it != m_SentPackets.end();) {
                auto seqn = (*it)->GetSeqn();
                if (seqn <= ackThrough) {
                    if (nackCount > 0) {
                        bool nacked = false;
                        for (int i = 0; i < nackCount; i++)
                            if (seqn == packet->GetNACK(i)) {
                                nacked = true;
                                break;
                            }
                        if (nacked) {
                            LogPrint(eLogDebug, "Streaming: Packet ", seqn, " NACK");
                            ++it;
                            continue;
                        }
                    }
                    auto sentPacket = *it;
                    uint64_t rtt = ts - sentPacket->sendTime;
                    if (ts < sentPacket->sendTime) {
                        LogPrint(eLogError, "Streaming: Packet ", seqn, "sent from the future, sendTime=",
                                 sentPacket->sendTime);
                        rtt = 1;
                    }
                    m_RTT = (m_RTT * seqn + rtt) / (seqn + 1);
                    m_RTO = m_RTT * 1.5; // TODO: implement it better
                    LogPrint(eLogDebug, "Streaming: Packet ", seqn, " acknowledged rtt=", rtt, " sentTime=",
                             sentPacket->sendTime);
                    m_SentPackets.erase(it++);
                    m_LocalDestination.DeletePacket(sentPacket);
                    acknowledged = true;
                    if (m_WindowSize < WINDOW_SIZE)
                        m_WindowSize++; // slow start
                    else {
                        // linear growth
                        if (ts > m_LastWindowSizeIncreaseTime + m_RTT) {
                            m_WindowSize++;
                            if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE;
                            m_LastWindowSizeIncreaseTime = ts;
                        }
                    }
                    if (!seqn && m_RoutingSession) // first message confirmed
                        m_RoutingSession->SetSharedRoutingPath(
                                std::make_shared<i2p::garlic::GarlicRoutingPath>(
                                        i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease,
                                                                       m_RTT, 0, 0}));
                } else
                    break;
            }
            if (m_SentPackets.empty())
                m_ResendTimer.cancel();
            if (acknowledged) {
                m_NumResendAttempts = 0;
                SendBuffer();
            }
            if (m_Status == eStreamStatusClosed)
                Terminate();
            else if (m_Status == eStreamStatusClosing)
                Close(); // check is all outgoing messages have been sent and we can send close
        }

        size_t Stream::Send(const uint8_t *buf, size_t len) {
            AsyncSend(buf, len, nullptr);
            return len;
        }

        void Stream::AsyncSend(const uint8_t *buf, size_t len, SendHandler handler) {
            if (len > 0 && buf) {
                std::unique_lock<std::mutex> l(m_SendBufferMutex);
                m_SendBuffer.Add(buf, len, handler);
            } else if (handler)
                handler(boost::system::error_code());
            m_Service.post(std::bind(&Stream::SendBuffer, shared_from_this()));
        }

        void Stream::SendBuffer() {
            int numMsgs = m_WindowSize - m_SentPackets.size();
            if (numMsgs <= 0) return; // window is full

            bool isNoAck = m_LastReceivedSequenceNumber < 0; // first packet
            std::vector<Packet *> packets;
            {
                std::unique_lock<std::mutex> l(m_SendBufferMutex);
                while ((m_Status == eStreamStatusNew) || (IsEstablished() && !m_SendBuffer.IsEmpty() && numMsgs > 0)) {
                    Packet *p = m_LocalDestination.NewPacket();
                    uint8_t *packet = p->GetBuffer();
                    // TODO: implement setters
                    size_t size = 0;
                    htobe32buf(packet + size, m_SendStreamID);
                    size += 4; // sendStreamID
                    htobe32buf(packet + size, m_RecvStreamID);
                    size += 4; // receiveStreamID
                    htobe32buf(packet + size, m_SequenceNumber++);
                    size += 4; // sequenceNum
                    if (isNoAck)
                        htobuf32(packet + size, 0);
                    else
                        htobe32buf(packet + size, m_LastReceivedSequenceNumber);
                    size += 4; // ack Through
                    packet[size] = 0;
                    size++; // NACK count
                    packet[size] = m_RTO / 1000;
                    size++; // resend delay
                    if (m_Status == eStreamStatusNew) {
                        // initial packet
                        m_Status = eStreamStatusOpen;
                        if (!m_RemoteLeaseSet)
                            m_RemoteLeaseSet = m_LocalDestination.GetOwner()->FindLeaseSet(
                                    m_RemoteIdentity->GetIdentHash());;
                        if (m_RemoteLeaseSet) {
                            m_RoutingSession = m_LocalDestination.GetOwner()->GetRoutingSession(m_RemoteLeaseSet, true);
                            m_MTU = m_RoutingSession->IsRatchets() ? STREAMING_MTU_RATCHETS : STREAMING_MTU;
                        }
                        uint16_t flags = PACKET_FLAG_SYNCHRONIZE | PACKET_FLAG_FROM_INCLUDED |
                                         PACKET_FLAG_SIGNATURE_INCLUDED | PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED;
                        if (isNoAck) flags |= PACKET_FLAG_NO_ACK;
                        bool isOfflineSignature = m_LocalDestination.GetOwner()->GetPrivateKeys().IsOfflineSignature();
                        if (isOfflineSignature) flags |= PACKET_FLAG_OFFLINE_SIGNATURE;
                        htobe16buf(packet + size, flags);
                        size += 2; // flags
                        size_t identityLen = m_LocalDestination.GetOwner()->GetIdentity()->GetFullLen();
                        size_t signatureLen = m_LocalDestination.GetOwner()->GetPrivateKeys().GetSignatureLen();
                        uint8_t *optionsSize = packet + size; // set options size later
                        size += 2; // options size
                        m_LocalDestination.GetOwner()->GetIdentity()->ToBuffer(packet + size, identityLen);
                        size += identityLen; // from
                        htobe16buf(packet + size, m_MTU);
                        size += 2; // max packet size
                        if (isOfflineSignature) {
                            const auto &offlineSignature = m_LocalDestination.GetOwner()->GetPrivateKeys().GetOfflineSignature();
                            memcpy(packet + size, offlineSignature.data(), offlineSignature.size());
                            size += offlineSignature.size(); // offline signature
                        }
                        uint8_t *signature = packet + size; // set it later
                        memset(signature, 0, signatureLen); // zeroes for now
                        size += signatureLen; // signature
                        htobe16buf(optionsSize, packet + size - 2 - optionsSize); // actual options size
                        size += m_SendBuffer.Get(packet + size, m_MTU); // payload
                        m_LocalDestination.GetOwner()->Sign(packet, size, signature);
                    } else {
                        // follow on packet
                        htobuf16(packet + size, 0);
                        size += 2; // flags
                        htobuf16(packet + size, 0); // no options
                        size += 2; // options size
                        size += m_SendBuffer.Get(packet + size, m_MTU); // payload
                    }
                    p->len = size;
                    packets.push_back(p);
                    numMsgs--;
                }
            }
            if (packets.size() > 0) {
                if (m_SavedPackets.empty()) // no NACKS
                {
                    m_IsAckSendScheduled = false;
                    m_AckSendTimer.cancel();
                }
                bool isEmpty = m_SentPackets.empty();
                auto ts = i2p::util::GetMillisecondsSinceEpoch();
                for (auto &it: packets) {
                    it->sendTime = ts;
                    m_SentPackets.insert(it);
                }
                SendPackets(packets);
                if (m_Status == eStreamStatusClosing && m_SendBuffer.IsEmpty())
                    SendClose();
                if (isEmpty)
                    ScheduleResend();
            }
        }

        void Stream::SendQuickAck() {
            int32_t lastReceivedSeqn = m_LastReceivedSequenceNumber;
            if (!m_SavedPackets.empty()) {
                int32_t seqn = (*m_SavedPackets.rbegin())->GetSeqn();
                if (seqn > lastReceivedSeqn) lastReceivedSeqn = seqn;
            }
            if (lastReceivedSeqn < 0) {
                LogPrint(eLogError, "Streaming: No packets have been received yet");
                return;
            }

            Packet p;
            uint8_t *packet = p.GetBuffer();
            size_t size = 0;
            htobe32buf(packet + size, m_SendStreamID);
            size += 4; // sendStreamID
            htobe32buf(packet + size, m_RecvStreamID);
            size += 4; // receiveStreamID
            htobuf32(packet + size, 0); // this is plain Ack message
            size += 4; // sequenceNum
            htobe32buf(packet + size, lastReceivedSeqn);
            size += 4; // ack Through
            uint8_t numNacks = 0;
            if (lastReceivedSeqn > m_LastReceivedSequenceNumber) {
                // fill NACKs
                uint8_t *nacks = packet + size + 1;
                auto nextSeqn = m_LastReceivedSequenceNumber + 1;
                for (auto it: m_SavedPackets) {
                    auto seqn = it->GetSeqn();
                    if (numNacks + (seqn - nextSeqn) >= 256) {
                        LogPrint(eLogError, "Streaming: Number of NACKs exceeds 256. seqn=", seqn, " nextSeqn=",
                                 nextSeqn);
                        htobe32buf(packet + 12, nextSeqn); // change ack Through
                        break;
                    }
                    for (uint32_t i = nextSeqn; i < seqn; i++) {
                        htobe32buf(nacks, i);
                        nacks += 4;
                        numNacks++;
                    }
                    nextSeqn = seqn + 1;
                }
                packet[size] = numNacks;
                size++; // NACK count
                size += numNacks * 4; // NACKs
            } else {
                // No NACKs
                packet[size] = 0;
                size++; // NACK count
            }
            packet[size] = 0;
            size++; // resend delay
            htobuf16(packet + size, 0); // no flags set
            size += 2; // flags
            htobuf16(packet + size, 0); // no options
            size += 2; // options size
            p.len = size;

            SendPackets(std::vector<Packet *>{&p});
            LogPrint(eLogDebug, "Streaming: Quick Ack sent. ", (int) numNacks, " NACKs");
        }

        void Stream::SendPing() {
            Packet p;
            uint8_t *packet = p.GetBuffer();
            size_t size = 0;
            htobe32buf(packet, m_RecvStreamID);
            size += 4; // sendStreamID
            memset(packet + size, 0, 14);
            size += 14; // all zeroes
            uint16_t flags = PACKET_FLAG_ECHO | PACKET_FLAG_SIGNATURE_INCLUDED | PACKET_FLAG_FROM_INCLUDED;
            bool isOfflineSignature = m_LocalDestination.GetOwner()->GetPrivateKeys().IsOfflineSignature();
            if (isOfflineSignature) flags |= PACKET_FLAG_OFFLINE_SIGNATURE;
            htobe16buf(packet + size, flags);
            size += 2; // flags
            size_t identityLen = m_LocalDestination.GetOwner()->GetIdentity()->GetFullLen();
            size_t signatureLen = m_LocalDestination.GetOwner()->GetPrivateKeys().GetSignatureLen();
            uint8_t *optionsSize = packet + size; // set options size later
            size += 2; // options size
            m_LocalDestination.GetOwner()->GetIdentity()->ToBuffer(packet + size, identityLen);
            size += identityLen; // from
            if (isOfflineSignature) {
                const auto &offlineSignature = m_LocalDestination.GetOwner()->GetPrivateKeys().GetOfflineSignature();
                memcpy(packet + size, offlineSignature.data(), offlineSignature.size());
                size += offlineSignature.size(); // offline signature
            }
            uint8_t *signature = packet + size; // set it later
            memset(signature, 0, signatureLen); // zeroes for now
            size += signatureLen; // signature
            htobe16buf(optionsSize, packet + size - 2 - optionsSize); // actual options size
            m_LocalDestination.GetOwner()->Sign(packet, size, signature);
            p.len = size;
            SendPackets(std::vector<Packet *>{&p});
            LogPrint(eLogDebug, "Streaming: Ping of ", p.len, " bytes sent");
        }

        void Stream::Close() {
            LogPrint(eLogDebug, "Streaming: closing stream with sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID,
                     ", status=", m_Status);
            switch (m_Status) {
                case eStreamStatusOpen:
                    m_Status = eStreamStatusClosing;
                    Close(); // recursion
                    if (m_Status == eStreamStatusClosing) //still closing
                        LogPrint(eLogDebug, "Streaming: Trying to send stream data before closing, sSID=",
                                 m_SendStreamID);
                    break;
                case eStreamStatusReset:
                    // TODO: send reset
                    Terminate();
                    break;
                case eStreamStatusClosing:
                    if (m_SentPackets.empty() && m_SendBuffer.IsEmpty()) // nothing to send
                    {
                        m_Status = eStreamStatusClosed;
                        SendClose();
                    }
                    break;
                case eStreamStatusClosed:
                    // already closed
                    Terminate();
                    break;
                default:
                    LogPrint(eLogWarning, "Streaming: Unexpected stream status=", (int) m_Status, " for sSID=",
                             m_SendStreamID);
            };
        }

        void Stream::SendClose() {
            Packet *p = m_LocalDestination.NewPacket();
            uint8_t *packet = p->GetBuffer();
            size_t size = 0;
            htobe32buf(packet + size, m_SendStreamID);
            size += 4; // sendStreamID
            htobe32buf(packet + size, m_RecvStreamID);
            size += 4; // receiveStreamID
            htobe32buf(packet + size, m_SequenceNumber++);
            size += 4; // sequenceNum
            htobe32buf(packet + size, m_LastReceivedSequenceNumber >= 0 ? m_LastReceivedSequenceNumber : 0);
            size += 4; // ack Through
            packet[size] = 0;
            size++; // NACK count
            packet[size] = 0;
            size++; // resend delay
            htobe16buf(packet + size, PACKET_FLAG_CLOSE | PACKET_FLAG_SIGNATURE_INCLUDED);
            size += 2; // flags
            size_t signatureLen = m_LocalDestination.GetOwner()->GetPrivateKeys().GetSignatureLen();
            htobe16buf(packet + size, signatureLen); // signature only
            size += 2; // options size
            uint8_t *signature = packet + size;
            memset(packet + size, 0, signatureLen);
            size += signatureLen; // signature
            m_LocalDestination.GetOwner()->Sign(packet, size, signature);

            p->len = size;
            m_Service.post(std::bind(&Stream::SendPacket, shared_from_this(), p));
            LogPrint(eLogDebug, "Streaming: FIN sent, sSID=", m_SendStreamID);
        }

        size_t Stream::ConcatenatePackets(uint8_t *buf, size_t len) {
            size_t pos = 0;
            while (pos < len && !m_ReceiveQueue.empty()) {
                Packet *packet = m_ReceiveQueue.front();
                size_t l = std::min(packet->GetLength(), len - pos);
                memcpy(buf + pos, packet->GetBuffer(), l);
                pos += l;
                packet->offset += l;
                if (!packet->GetLength()) {
                    m_ReceiveQueue.pop();
                    m_LocalDestination.DeletePacket(packet);
                }
            }
            return pos;
        }

        bool Stream::SendPacket(Packet *packet) {
            if (packet) {
                if (m_IsAckSendScheduled) {
                    m_IsAckSendScheduled = false;
                    m_AckSendTimer.cancel();
                }
                SendPackets(std::vector<Packet *>{packet});
                bool isEmpty = m_SentPackets.empty();
                m_SentPackets.insert(packet);
                if (isEmpty)
                    ScheduleResend();
                return true;
            } else
                return false;
        }

        void Stream::SendPackets(const std::vector<Packet *> &packets) {
            if (!m_RemoteLeaseSet) {
                UpdateCurrentRemoteLease();
                if (!m_RemoteLeaseSet) {
                    LogPrint(eLogError, "Streaming: Can't send packets, missing remote LeaseSet, sSID=",
                             m_SendStreamID);
                    return;
                }
            }
            if (!m_RoutingSession || m_RoutingSession->IsTerminated() ||
                !m_RoutingSession->IsReadyToSend()) // expired and detached or new session sent
                m_RoutingSession = m_LocalDestination.GetOwner()->GetRoutingSession(m_RemoteLeaseSet, true);
            if (!m_CurrentOutboundTunnel && m_RoutingSession) // first message to send
            {
                // try to get shared path first
                auto routingPath = m_RoutingSession->GetSharedRoutingPath();
                if (routingPath) {
                    m_CurrentOutboundTunnel = routingPath->outboundTunnel;
                    m_CurrentRemoteLease = routingPath->remoteLease;
                    m_RTT = routingPath->rtt;
                    m_RTO = m_RTT * 1.5; // TODO: implement it better
                }
            }

            auto ts = i2p::util::GetMillisecondsSinceEpoch();
            if (!m_CurrentRemoteLease || !m_CurrentRemoteLease->endDate || // excluded from LeaseSet
                ts >= m_CurrentRemoteLease->endDate - i2p::data::LEASE_ENDDATE_THRESHOLD)
                UpdateCurrentRemoteLease(true);
            if (m_CurrentRemoteLease && ts < m_CurrentRemoteLease->endDate + i2p::data::LEASE_ENDDATE_THRESHOLD) {
                if (!m_CurrentOutboundTunnel) {
                    auto leaseRouter = i2p::data::netdb.FindRouter(m_CurrentRemoteLease->tunnelGateway);
                    m_CurrentOutboundTunnel = m_LocalDestination.GetOwner()->GetTunnelPool()->GetNextOutboundTunnel(
                            nullptr,
                            leaseRouter ? leaseRouter->GetCompatibleTransports(false)
                                        : (i2p::data::RouterInfo::CompatibleTransports) i2p::data::RouterInfo::eAllTransports);
                } else if (!m_CurrentOutboundTunnel->IsEstablished())
                    m_CurrentOutboundTunnel = m_LocalDestination.GetOwner()->GetTunnelPool()->GetNewOutboundTunnel(
                            m_CurrentOutboundTunnel);
                if (!m_CurrentOutboundTunnel) {
                    LogPrint(eLogError, "Streaming: No outbound tunnels in the pool, sSID=", m_SendStreamID);
                    m_CurrentRemoteLease = nullptr;
                    return;
                }

                std::vector<i2p::tunnel::TunnelMessageBlock> msgs;
                for (const auto &it: packets) {
                    auto msg = m_RoutingSession->WrapSingleMessage(m_LocalDestination.CreateDataMessage(
                            it->GetBuffer(), it->GetLength(), m_Port, !m_RoutingSession->IsRatchets()));
                    msgs.push_back(i2p::tunnel::TunnelMessageBlock
                                           {
                                                   i2p::tunnel::eDeliveryTypeTunnel,
                                                   m_CurrentRemoteLease->tunnelGateway, m_CurrentRemoteLease->tunnelID,
                                                   msg
                                           });
                    m_NumSentBytes += it->GetLength();
                }
                m_CurrentOutboundTunnel->SendTunnelDataMsg(msgs);
            } else {
                LogPrint(eLogWarning, "Streaming: Remote lease is not available, sSID=", m_SendStreamID);
                if (m_RoutingSession)
                    m_RoutingSession->SetSharedRoutingPath(nullptr); // invalidate routing path
            }
        }

        void Stream::SendUpdatedLeaseSet() {
            if (m_RoutingSession && !m_RoutingSession->IsTerminated()) {
                if (m_RoutingSession->IsLeaseSetNonConfirmed()) {
                    auto ts = i2p::util::GetMillisecondsSinceEpoch();
                    if (ts > m_RoutingSession->GetLeaseSetSubmissionTime() + i2p::garlic::LEASET_CONFIRMATION_TIMEOUT) {
                        // LeaseSet was not confirmed, should try other tunnels
                        LogPrint(eLogWarning, "Streaming: LeaseSet was not confirmed in ",
                                 i2p::garlic::LEASET_CONFIRMATION_TIMEOUT, " milliseconds. Trying to resubmit");
                        m_RoutingSession->SetSharedRoutingPath(nullptr);
                        m_CurrentOutboundTunnel = nullptr;
                        m_CurrentRemoteLease = nullptr;
                        SendQuickAck();
                    }
                } else if (m_RoutingSession->IsLeaseSetUpdated()) {
                    LogPrint(eLogDebug, "Streaming: sending updated LeaseSet");
                    SendQuickAck();
                }
            } else
                SendQuickAck();
        }

        void Stream::ScheduleResend() {
            if (m_Status != eStreamStatusTerminated) {
                m_ResendTimer.cancel();
                // check for invalid value
                if (m_RTO <= 0) m_RTO = INITIAL_RTO;
                m_ResendTimer.expires_from_now(boost::posix_time::milliseconds(m_RTO));
                m_ResendTimer.async_wait(std::bind(&Stream::HandleResendTimer,
                                                   shared_from_this(), std::placeholders::_1));
            }
        }

        void Stream::HandleResendTimer(const boost::system::error_code &ecode) {
            if (ecode != boost::asio::error::operation_aborted) {
                // check for resend attempts
                if (m_NumResendAttempts >= MAX_NUM_RESEND_ATTEMPTS) {
                    LogPrint(eLogWarning, "Streaming: packet was not ACKed after ", MAX_NUM_RESEND_ATTEMPTS,
                             " attempts, terminate, rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID);
                    m_Status = eStreamStatusReset;
                    Close();
                    return;
                }

                // collect packets to resend
                auto ts = i2p::util::GetMillisecondsSinceEpoch();
                std::vector<Packet *> packets;
                for (auto it: m_SentPackets) {
                    if (ts >= it->sendTime + m_RTO) {
                        it->sendTime = ts;
                        packets.push_back(it);
                    }
                }

                // select tunnels if necessary and send
                if (packets.size() > 0) {
                    m_NumResendAttempts++;
                    m_RTO *= 2;
                    switch (m_NumResendAttempts) {
                        case 1: // congesion avoidance
                            m_WindowSize >>= 1; // /2
                            if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
                            break;
                        case 2:
                            m_RTO = INITIAL_RTO; // drop RTO to initial upon tunnels pair change first time
#if (__cplusplus >= 201703L) // C++ 17 or higher
                            [[fallthrough]];
#endif
                            // no break here
                        case 4:
                            if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath(nullptr);
                            UpdateCurrentRemoteLease(); // pick another lease
                            LogPrint(eLogWarning,
                                     "Streaming: Another remote lease has been selected for stream with rSID=",
                                     m_RecvStreamID, ", sSID=", m_SendStreamID);
                            break;
                        case 3:
                            // pick another outbound tunnel
                            if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath(nullptr);
                            m_CurrentOutboundTunnel = m_LocalDestination.GetOwner()->GetTunnelPool()->GetNextOutboundTunnel(
                                    m_CurrentOutboundTunnel);
                            LogPrint(eLogWarning,
                                     "Streaming: Another outbound tunnel has been selected for stream with sSID=",
                                     m_SendStreamID);
                            break;
                        default:;
                    }
                    SendPackets(packets);
                }
                ScheduleResend();
            }
        }

        void Stream::HandleAckSendTimer(const boost::system::error_code &ecode) {
            if (m_IsAckSendScheduled) {
                if (m_LastReceivedSequenceNumber < 0) {
                    LogPrint(eLogWarning, "Streaming: SYN has not been received after ", SYN_TIMEOUT,
                             " milliseconds after follow on, terminate rSID=", m_RecvStreamID, ", sSID=",
                             m_SendStreamID);
                    m_Status = eStreamStatusReset;
                    Close();
                    return;
                }
                if (m_Status == eStreamStatusOpen) {
                    if (m_RoutingSession && m_RoutingSession->IsLeaseSetNonConfirmed()) {
                        // seems something went wrong and we should re-select tunnels
                        m_CurrentOutboundTunnel = nullptr;
                        m_CurrentRemoteLease = nullptr;
                    }
                    SendQuickAck();
                }
                m_IsAckSendScheduled = false;
            }
        }

        void Stream::UpdateCurrentRemoteLease(bool expired) {
            if (!m_RemoteLeaseSet || m_RemoteLeaseSet->IsExpired()) {
                auto remoteLeaseSet = m_LocalDestination.GetOwner()->FindLeaseSet(m_RemoteIdentity->GetIdentHash());
                if (!remoteLeaseSet) {
                    LogPrint(eLogWarning, "Streaming: LeaseSet ", m_RemoteIdentity->GetIdentHash().ToBase64(),
                             m_RemoteLeaseSet ? " expired" : " not found");
                    if (m_RemoteLeaseSet && m_RemoteLeaseSet->IsPublishedEncrypted()) {
                        m_LocalDestination.GetOwner()->RequestDestinationWithEncryptedLeaseSet(
                                std::make_shared<i2p::data::BlindedPublicKey>(m_RemoteIdentity));
                        return; // we keep m_RemoteLeaseSet for possible next request
                    } else {
                        m_RemoteLeaseSet = nullptr;
                        m_LocalDestination.GetOwner()->RequestDestination(
                                m_RemoteIdentity->GetIdentHash()); // try to request for a next attempt
                    }
                } else {
                    // LeaseSet updated
                    m_RemoteLeaseSet = remoteLeaseSet;
                    m_RemoteIdentity = m_RemoteLeaseSet->GetIdentity();
                    m_TransientVerifier = m_RemoteLeaseSet->GetTransientVerifier();
                }
            }
            if (m_RemoteLeaseSet) {
                if (!m_RoutingSession)
                    m_RoutingSession = m_LocalDestination.GetOwner()->GetRoutingSession(m_RemoteLeaseSet, true);
                auto leases = m_RemoteLeaseSet->GetNonExpiredLeases(false); // try without threshold first
                if (leases.empty()) {
                    expired = false;
                    // time to request
                    if (m_RemoteLeaseSet->IsPublishedEncrypted())
                        m_LocalDestination.GetOwner()->RequestDestinationWithEncryptedLeaseSet(
                                std::make_shared<i2p::data::BlindedPublicKey>(m_RemoteIdentity));
                    else
                        m_LocalDestination.GetOwner()->RequestDestination(m_RemoteIdentity->GetIdentHash());
                    leases = m_RemoteLeaseSet->GetNonExpiredLeases(true); // then with threshold
                }
                if (!leases.empty()) {
                    bool updated = false;
                    if (expired && m_CurrentRemoteLease) {
                        for (const auto &it: leases)
                            if ((it->tunnelGateway == m_CurrentRemoteLease->tunnelGateway) &&
                                (it->tunnelID != m_CurrentRemoteLease->tunnelID)) {
                                m_CurrentRemoteLease = it;
                                updated = true;
                                break;
                            }
                    }
                    if (!updated) {
                        uint32_t i = rand() % leases.size();
                        if (m_CurrentRemoteLease && leases[i]->tunnelID == m_CurrentRemoteLease->tunnelID)
                            // make sure we don't select previous
                            i = (i + 1) % leases.size(); // if so, pick next
                        m_CurrentRemoteLease = leases[i];
                    }
                } else {
                    LogPrint(eLogWarning, "Streaming: All remote leases are expired");
                    m_RemoteLeaseSet = nullptr;
                    m_CurrentRemoteLease = nullptr;
                    // we have requested expired before, no need to do it twice
                }
            } else {
                LogPrint(eLogWarning, "Streaming: Remote LeaseSet not found");
                m_CurrentRemoteLease = nullptr;
            }
        }

        StreamingDestination::StreamingDestination(std::shared_ptr<i2p::client::ClientDestination> owner,
                                                   uint16_t localPort, bool gzip) :
                m_Owner(owner), m_LocalPort(localPort), m_Gzip(gzip),
                m_PendingIncomingTimer(m_Owner->GetService()) {
            if (m_Gzip)
                m_Deflator.reset(new i2p::data::GzipDeflator);
        }

        StreamingDestination::~StreamingDestination() {
            for (auto &it: m_SavedPackets) {
                for (auto it1: it.second) DeletePacket(it1);
                it.second.clear();
            }
            m_SavedPackets.clear();
        }

        void StreamingDestination::Start() {
        }

        void StreamingDestination::Stop() {
            ResetAcceptor();
            m_PendingIncomingTimer.cancel();
            m_PendingIncomingStreams.clear();
            {
                std::unique_lock<std::mutex> l(m_StreamsMutex);
                for (auto it: m_Streams)
                    it.second->Terminate(false); // we delete here
                m_Streams.clear();
                m_IncomingStreams.clear();
                m_LastStream = nullptr;
            }
        }

        void StreamingDestination::HandleNextPacket(Packet *packet) {
            uint32_t sendStreamID = packet->GetSendStreamID();
            if (sendStreamID) {
                if (!m_LastStream || sendStreamID != m_LastStream->GetRecvStreamID()) {
                    auto it = m_Streams.find(sendStreamID);
                    if (it != m_Streams.end())
                        m_LastStream = it->second;
                    else
                        m_LastStream = nullptr;
                }
                if (m_LastStream)
                    m_LastStream->HandleNextPacket(packet);
                else if (packet->IsEcho() && m_Owner->IsStreamingAnswerPings()) {
                    // ping
                    LogPrint(eLogInfo, "Streaming: Ping received sSID=", sendStreamID);
                    auto s = std::make_shared<Stream>(m_Owner->GetService(), *this);
                    s->HandlePing(packet);
                } else {
                    LogPrint(eLogInfo, "Streaming: Unknown stream sSID=", sendStreamID);
                    DeletePacket(packet);
                }
            } else {
                if (packet->IsEcho()) {
                    // pong
                    LogPrint(eLogInfo, "Streaming: Pong received rSID=", packet->GetReceiveStreamID());
                    DeletePacket(packet);
                    return;
                }
                if (packet->IsSYN() && !packet->GetSeqn()) // new incoming stream
                {
                    uint32_t receiveStreamID = packet->GetReceiveStreamID();
                    auto it1 = m_IncomingStreams.find(receiveStreamID);
                    if (it1 != m_IncomingStreams.end()) {
                        // already pending
                        LogPrint(eLogWarning, "Streaming: Incoming streaming with rSID=", receiveStreamID,
                                 " already exists");
                        DeletePacket(packet); // drop it, because previous should be connected
                        return;
                    }
                    auto incomingStream = CreateNewIncomingStream(receiveStreamID);
                    incomingStream->HandleNextPacket(packet); // SYN
                    auto ident = incomingStream->GetRemoteIdentity();

                    // handle saved packets if any
                    {
                        auto it = m_SavedPackets.find(receiveStreamID);
                        if (it != m_SavedPackets.end()) {
                            LogPrint(eLogDebug, "Streaming: Processing ", it->second.size(), " saved packets for rSID=",
                                     receiveStreamID);
                            for (auto it1: it->second)
                                incomingStream->HandleNextPacket(it1);
                            m_SavedPackets.erase(it);
                        }
                    }
                    // accept
                    if (m_Acceptor != nullptr)
                        m_Acceptor(incomingStream);
                    else {
                        LogPrint(eLogWarning, "Streaming: Acceptor for incoming stream is not set");
                        if (m_PendingIncomingStreams.size() < MAX_PENDING_INCOMING_BACKLOG) {
                            m_PendingIncomingStreams.push_back(incomingStream);
                            m_PendingIncomingTimer.cancel();
                            m_PendingIncomingTimer.expires_from_now(
                                    boost::posix_time::seconds(PENDING_INCOMING_TIMEOUT));
                            m_PendingIncomingTimer.async_wait(
                                    std::bind(&StreamingDestination::HandlePendingIncomingTimer,
                                              shared_from_this(), std::placeholders::_1));
                            LogPrint(eLogDebug, "Streaming: Pending incoming stream added, rSID=", receiveStreamID);
                        } else {
                            LogPrint(eLogWarning, "Streaming: Pending incoming streams backlog exceeds ",
                                     MAX_PENDING_INCOMING_BACKLOG);
                            incomingStream->Close();
                        }
                    }
                } else // follow on packet without SYN
                {
                    uint32_t receiveStreamID = packet->GetReceiveStreamID();
                    auto it1 = m_IncomingStreams.find(receiveStreamID);
                    if (it1 != m_IncomingStreams.end()) {
                        // found
                        it1->second->HandleNextPacket(packet);
                        return;
                    }
                    // save follow on packet
                    auto it = m_SavedPackets.find(receiveStreamID);
                    if (it != m_SavedPackets.end())
                        it->second.push_back(packet);
                    else {
                        m_SavedPackets[receiveStreamID] = std::list<Packet *>{packet};
                        auto timer = std::make_shared<boost::asio::deadline_timer>(m_Owner->GetService());
                        timer->expires_from_now(boost::posix_time::seconds(PENDING_INCOMING_TIMEOUT));
                        auto s = shared_from_this();
                        timer->async_wait([s, timer, receiveStreamID](const boost::system::error_code &ecode) {
                            if (ecode != boost::asio::error::operation_aborted) {
                                auto it = s->m_SavedPackets.find(receiveStreamID);
                                if (it != s->m_SavedPackets.end()) {
                                    for (auto it1: it->second) s->DeletePacket(it1);
                                    it->second.clear();
                                    s->m_SavedPackets.erase(it);
                                }
                            }
                        });
                    }
                }
            }
        }

        std::shared_ptr<Stream>
        StreamingDestination::CreateNewOutgoingStream(std::shared_ptr<const i2p::data::LeaseSet> remote, int port) {
            auto s = std::make_shared<Stream>(m_Owner->GetService(), *this, remote, port);
            std::unique_lock<std::mutex> l(m_StreamsMutex);
            m_Streams.emplace(s->GetRecvStreamID(), s);
            return s;
        }

        void StreamingDestination::SendPing(std::shared_ptr<const i2p::data::LeaseSet> remote) {
            auto s = std::make_shared<Stream>(m_Owner->GetService(), *this, remote, 0);
            s->SendPing();
        }

        std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream(uint32_t receiveStreamID) {
            auto s = std::make_shared<Stream>(m_Owner->GetService(), *this);
            std::unique_lock<std::mutex> l(m_StreamsMutex);
            m_Streams.emplace(s->GetRecvStreamID(), s);
            m_IncomingStreams.emplace(receiveStreamID, s);
            return s;
        }

        void StreamingDestination::DeleteStream(std::shared_ptr<Stream> stream) {
            if (stream) {
                std::unique_lock<std::mutex> l(m_StreamsMutex);
                m_Streams.erase(stream->GetRecvStreamID());
                m_IncomingStreams.erase(stream->GetSendStreamID());
                if (m_LastStream == stream) m_LastStream = nullptr;
            }
            if (m_Streams.empty()) {
                m_PacketsPool.CleanUp();
                m_I2NPMsgsPool.CleanUp();
            }
        }

        bool StreamingDestination::DeleteStream(uint32_t recvStreamID) {
            auto it = m_Streams.find(recvStreamID);
            if (it == m_Streams.end())
                return false;
            auto s = it->second;
            m_Owner->GetService().post([this, s]() {
                s->Close(); // try to send FIN
                s->Terminate(false);
                DeleteStream(s);
            });
            return true;
        }

        void StreamingDestination::SetAcceptor(const Acceptor &acceptor) {
            m_Acceptor = acceptor; // we must set it immediately for IsAcceptorSet
            auto s = shared_from_this();
            m_Owner->GetService().post([s](void) {
                // take care about incoming queue
                for (auto &it: s->m_PendingIncomingStreams)
                    if (it->GetStatus() == eStreamStatusOpen) // still open?
                        s->m_Acceptor(it);
                s->m_PendingIncomingStreams.clear();
                s->m_PendingIncomingTimer.cancel();
            });
        }

        void StreamingDestination::ResetAcceptor() {
            if (m_Acceptor) m_Acceptor(nullptr);
            m_Acceptor = nullptr;
        }

        void StreamingDestination::AcceptOnce(const Acceptor &acceptor) {
            m_Owner->GetService().post([acceptor, this](void) {
                if (!m_PendingIncomingStreams.empty()) {
                    acceptor(m_PendingIncomingStreams.front());
                    m_PendingIncomingStreams.pop_front();
                    if (m_PendingIncomingStreams.empty())
                        m_PendingIncomingTimer.cancel();
                } else // we must save old acceptor and set it back
                {
                    m_Acceptor = std::bind(&StreamingDestination::AcceptOnceAcceptor, this,
                                           std::placeholders::_1, acceptor, m_Acceptor);
                }
            });
        }

        void
        StreamingDestination::AcceptOnceAcceptor(std::shared_ptr<Stream> stream, Acceptor acceptor, Acceptor prev) {
            m_Acceptor = prev;
            acceptor(stream);
        }

        void StreamingDestination::HandlePendingIncomingTimer(const boost::system::error_code &ecode) {
            if (ecode != boost::asio::error::operation_aborted) {
                LogPrint(eLogWarning, "Streaming: Pending incoming timeout expired");
                for (auto &it: m_PendingIncomingStreams)
                    it->Close();
                m_PendingIncomingStreams.clear();
            }
        }

        void StreamingDestination::HandleDataMessagePayload(const uint8_t *buf, size_t len) {
            // unzip it
            Packet *uncompressed = NewPacket();
            uncompressed->offset = 0;
            uncompressed->len = m_Inflator.Inflate(buf, len, uncompressed->buf, MAX_PACKET_SIZE);
            if (uncompressed->len)
                HandleNextPacket(uncompressed);
            else
                DeletePacket(uncompressed);
        }

        std::shared_ptr<I2NPMessage> StreamingDestination::CreateDataMessage(
                const uint8_t *payload, size_t len, uint16_t toPort, bool checksum) {
            size_t size;
            auto msg = m_I2NPMsgsPool.AcquireShared();
            uint8_t *buf = msg->GetPayload();
            buf += 4; // reserve for lengthlength
            msg->len += 4;

            if (m_Gzip && m_Deflator)
                size = m_Deflator->Deflate(payload, len, buf, msg->maxLen - msg->len);
            else
                size = i2p::data::GzipNoCompression(payload, len, buf, msg->maxLen - msg->len);

            if (size) {
                htobe32buf(msg->GetPayload(), size); // length
                htobe16buf(buf + 4, m_LocalPort); // source port
                htobe16buf(buf + 6, toPort); // destination port
                buf[9] = i2p::client::PROTOCOL_TYPE_STREAMING; // streaming protocol
                msg->len += size;
                msg->FillI2NPMessageHeader(eI2NPData, 0, checksum);
            } else
                msg = nullptr;
            return msg;
        }

    }
}