receiver.hpp

// A class for handling limit order book depth messages from multi-cast.
// Copyright 2020 Christian Kauten
//
// Author: Christian Kauten (kautenja@auburn.edu)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU 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 General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
//
 
#ifndef DATA_FEED_RECEIVER_HPP
#define DATA_FEED_RECEIVER_HPP
 
#include <functional>
#include <iostream>
#include <asio.hpp>
#include "exceptions.hpp"
#include "limit_order_book/limit_order_book.hpp"
#include "messages.hpp"
 
namespace DataFeed {
 
template<typename Handler>
class Receiver {
 private:
    asio::ip::udp::socket socket;
    asio::ip::udp::endpoint sender_endpoint;
    Messages::PacketQueue input_buffer;
    LOB::LimitOrderBook book;
    bool _is_session_active = false;
    Handler& handler;
    SequenceNumber last_sequence = 0;
 
    void read_message() {
        input_buffer.emplace_back();
        socket.async_receive_from(
            asio::buffer(input_buffer.back(), sizeof(Messages::Packet)),
            sender_endpoint,
            [this](const std::error_code& error, std::size_t) {
                if (error)  // an error occurred while reading a message
                    throw Exception("DataFeed::read_message - " + error.message());
                // copy the message from the front of the queue and pop it off
                auto message = input_buffer.front();
                input_buffer.pop_front();
                // check if the sequence number is out of order
                if (reinterpret_cast<Messages::Header&>(message).sequence != ++last_sequence) {
                    std::cerr << "DataFeed::Receiver::read_message - out of sequence with server" << std::endl;
                    std::cerr << "DataFeed::Receiver::read_message - got sequence: " << reinterpret_cast<Messages::Header&>(message).sequence << std::endl;
                    std::cerr << "DataFeed::Receiver::read_message - expected sequence: " << last_sequence - 1 << std::endl;
                    last_sequence = reinterpret_cast<Messages::Header&>(message).sequence;
                }
                // read another message from the server
                read_message();
                // unwrap and handle the data using the template ID
                switch (reinterpret_cast<Messages::Header&>(message).uid) {
                    case Messages::MessageID::StartOfSession: { handle(reinterpret_cast<Messages::StartOfSession&>(message));  break; }
                    case Messages::MessageID::EndOfSession:   { handle(reinterpret_cast<Messages::EndOfSession&>(message));    break; }
                    case Messages::MessageID::Clear:          { handle(reinterpret_cast<Messages::Clear&>(message));           break; }
                    case Messages::MessageID::AddOrder:       { handle(reinterpret_cast<Messages::AddOrder&>(message));        break; }
                    case Messages::MessageID::DeleteOrder:    { handle(reinterpret_cast<Messages::DeleteOrder&>(message));     break; }
                    case Messages::MessageID::Trade:          { handle(reinterpret_cast<Messages::Trade&>(message));           break; }
                    default:                                  { handle(reinterpret_cast<Messages::Header&>(message));          break; }
                }
            }
        );
    }
 
    inline void handle(const Messages::StartOfSession& message) {
        // set the trading session to active
        _is_session_active = true;
        // pass the message to the handler
        handler.did_receive(this, message);
    }
 
    inline void handle(const Messages::EndOfSession& message) {
        // set the trading session to inactive
        _is_session_active = false;
        // pass the message to the handler
        handler.did_receive(this, message);
    }
 
    inline void handle(const Messages::Clear& message) {
        // clear all the orders in the book
        book.clear();
        // pass the message to the handler
        handler.did_receive(this, message);
    }
 
    inline void handle(const Messages::AddOrder& message) {
        // add the order to the book
        book.limit(side_to_LOB_side(message.side), message.uid, message.quantity, message.price);
        // pass the message to the handler
        handler.did_receive(this, message);
    }
 
    inline void handle(const Messages::DeleteOrder& message) {
        if (not book.has(message.uid)) {  // the limit order was not found
            std::cerr << "DataFeed::Receiver::handle(DeleteOrder) - received delete for non-existent order: " << message.uid << std::endl;
            // TODO: handle error
            return;
        }
        // delete the order in the book
        book.cancel(message.uid);
        // pass the message to the handler
        handler.did_receive(this, message);
    }
 
    inline void handle(const Messages::Trade& message) {
        if (not book.has(message.uid)) {  // the limit order was not found
            std::cerr << "DataFeed::Receiver::handle(Trade) - received trade for non-existent order: " << message.uid << std::endl;
            // TODO: handle error
            return;
        }
        // reduce the size of the given order
        // (removes order if remaining size is 0)
        book.reduce(message.uid, message.quantity);
        // pass the message to the handler
        handler.did_receive(this, message);
    }
 
    inline void handle(const Messages::Header& header) {
        throw Exception("Receiver::handle(Header) - received message with invalid header " + header.to_string());
    }
 
 public:
    Receiver(
        asio::io_context& io_context,
        const asio::ip::address& listen,
        const asio::ip::address& group,
        uint16_t port,
        Handler& handler_,
        bool reuse_enabled = true
    ) : socket(io_context), handler(handler_) {
        // Create the socket so that multiple may be bound to the same address.
        asio::ip::udp::endpoint listen_endpoint(listen, port);
        socket.open(listen_endpoint.protocol());
        socket.set_option(asio::ip::udp::socket::reuse_address(reuse_enabled));
        socket.bind(asio::ip::udp::endpoint(group, port));
        socket.set_option(asio::ip::multicast::join_group(group));
        // detect if the compilation platform is MacOS and don't try to set
        // the buffer, if so.
#if __APPLE__
#else
        // set the size of the receive buffer to the maximum allowed size
        auto size = std::numeric_limits<uint32_t>::max();
        auto recv_buffer_size = asio::socket_base::receive_buffer_size(size);
        socket.set_option(recv_buffer_size);
#endif
        // read the first message
        read_message();
    }
 
    inline const LOB::LimitOrderBook& get_book() { return book; }
 
    inline bool is_session_active() const { return _is_session_active; }
 
    inline const Handler& get_handler() const { return handler; }
};
 
}    // namespace DataFeed
 
#endif    // DATA_FEED_RECEIVER_HPP