path: root/src/Net/Connection.h

/*
 * Connection.h
 *
 * Copyright (C) 2008 Matthias Schiffer <matthias@gamezock.de>
 *
 * 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 MAD_NET_CONNECTION_H_
#define MAD_NET_CONNECTION_H_

#include "export.h"

#include "Packet.h"
#include <Core/Signals.h>
#include <Core/ThreadManager.h>

#include <boost/array.hpp>
#include <boost/shared_array.hpp>

#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>

#include <boost/thread/shared_mutex.hpp>

namespace Mad {
namespace Net {

class Listener;

class MAD_NET_EXPORT Connection : boost::noncopyable {
  protected:
    friend class Listener;

    enum State {
      DISCONNECTED, HANDSHAKE, CONNECT, CONNECTED, DISCONNECT, SHUTDOWN
    };

  private:
    class MAD_NET_EXPORT Buffer {
      public:
        Buffer(const boost::uint8_t *data0, std::size_t length) : data(new std::vector<boost::uint8_t>(data0, data0+length)), buffer(boost::asio::buffer(*data)) {}

        typedef boost::asio::const_buffer value_type;
        typedef const boost::asio::const_buffer* const_iterator;

        const boost::asio::const_buffer* begin() const { return &buffer; }
        const boost::asio::const_buffer* end() const { return &buffer + 1; }

      private:
        boost::shared_ptr<std::vector<boost::uint8_t> > data;
        boost::asio::const_buffer buffer;
    };

    Core::Application *application;

    boost::condition_variable_any stateChanged;

    State state;

    bool dontStart;

    boost::scoped_ptr<boost::array<boost::uint8_t, 1024*1024> > receiveBuffer;
    std::size_t received;

    Packet::Header header;

    Core::Signals::Signal1<boost::shared_ptr<Packet> > receiveSignal;
    Core::Signals::Signal0 connectedSignal;
    Core::Signals::Signal0 disconnectedSignal;

    boost::uint32_t receiveLimit;

    bool receiving;
    unsigned long sending;

    void enterReceiveLoop();

    void handleHeaderReceive(const boost::shared_array<boost::uint8_t> &data);
    void handleDataReceive(const boost::shared_array<boost::uint8_t> &data);

    void handleRead(const boost::system::error_code& error, std::size_t bytes_transferred, std::size_t length, const boost::function1<void, const boost::shared_array<boost::uint8_t>& > &notify);
    void handleWrite(const boost::system::error_code& error, std::size_t);

    void handleShutdown(const boost::system::error_code& error);

    void rawReceive(std::size_t length, const boost::function1<void, const boost::shared_array<boost::uint8_t>& > &notify);
    void rawSend(const boost::uint8_t *data, std::size_t length);

  protected:
    boost::weak_ptr<Connection> thisPtr;

    boost::shared_mutex connectionLock;
    
    boost::shared_ptr<boost::asio::ssl::context> context;
    boost::asio::ssl::stream<boost::asio::ip::tcp::socket> socket;
    boost::asio::ip::tcp::endpoint peer;

    void handleHandshake(const boost::system::error_code& error);

    bool _isConnected() const {return (state != DISCONNECTED);}
    bool _isConnecting() const {
      return (state == HANDSHAKE || state == CONNECT);
    }

    bool _isDisconnecting() const {
      return (state == DISCONNECT || state == SHUTDOWN);
    }

    void _setState(State newState) {
      state = newState;
      
      stateChanged.notify_all();
    }

    void doDisconnect() {
      boost::unique_lock<boost::shared_mutex> lock(connectionLock);

      if(_isConnected() && state != SHUTDOWN) {
        _setState(SHUTDOWN);
        boost::system::error_code error;
        socket.lowest_layer().cancel(error);

        socket.async_shutdown(boost::bind(&Connection::handleShutdown, thisPtr.lock(), boost::asio::placeholders::error));
      }
    }

    Connection(Core::Application *application0, boost::shared_ptr<boost::asio::ssl::context> context0) :
      application(application0), state(DISCONNECTED), dontStart(false),
      receiveBuffer(new boost::array<boost::uint8_t, 1024*1024>),
      receiveSignal(application), connectedSignal(application),
      disconnectedSignal(application), receiveLimit(0xFFFF) /* 64K */, receiving(false), sending(0),
      context(context0), socket(application->getIOService(), *context) {}

    static boost::shared_ptr<Connection> create(Core::Application *application, boost::shared_ptr<boost::asio::ssl::context> context) {
      boost::shared_ptr<Connection> connection(new Connection(application, context));

      connection->thisPtr = connection;

      return connection;
    }

  public:
    virtual ~Connection();

    bool isConnected() {
      boost::shared_lock<boost::shared_mutex> lock(connectionLock);
      return _isConnected();
    }

    bool isConnecting() {
      boost::shared_lock<boost::shared_mutex> lock(connectionLock);
      return _isConnecting();
    }

    bool isDisconnecting() {
      boost::shared_lock<boost::shared_mutex> lock(connectionLock);
      return _isDisconnecting();
    }

    void waitWhileConnecting() {
      boost::shared_lock<boost::shared_mutex> lock(connectionLock);

      while(_isConnecting())
        stateChanged.wait(lock);
    }

    void waitWhileConnected() {
      boost::shared_lock<boost::shared_mutex> lock(connectionLock);

      while(_isConnected())
        stateChanged.wait(lock);
    }

    boost::asio::ip::tcp::endpoint getPeer() {
      boost::shared_lock<boost::shared_mutex> lock(connectionLock);
      return peer;
    }

    void setStart(bool start = true) {
      boost::lock_guard<boost::shared_mutex> lock(connectionLock);

      dontStart = !start;
    }

    void unsetStart() {
      setStart(false);
    }

    boost::uint32_t getReceiveLimit() {
      boost::shared_lock<boost::shared_mutex> lock(connectionLock);
      return receiveLimit;
    }

    void setReceiveLimit(boost::uint32_t limit) {
      boost::lock_guard<boost::shared_mutex> lock(connectionLock);
      receiveLimit = limit;
    }

    void startReceive() {
      {
        boost::lock_guard<boost::shared_mutex> lock(connectionLock);

        if(state != CONNECT)
          return;

        _setState(CONNECTED);
      }

      enterReceiveLoop();
    }

    void disconnect();

    bool send(const Packet &packet);

    Core::Signals::Connection connectSignalReceive(const Core::Signals::Signal1<boost::shared_ptr<Packet> >::slot_type &slot) {
      return receiveSignal.connect(slot);
    }
    void disconnectSignalReceive(const Core::Signals::Connection &connection) {
      receiveSignal.disconnect(connection);
    }

    Core::Signals::Connection connectSignalConnected(const Core::Signals::Signal0::slot_type &slot) {
      return connectedSignal.connect(slot);
    }
    void disconnectSignalConnected(const Core::Signals::Connection &connection) {
      connectedSignal.disconnect(connection);
    }

    Core::Signals::Connection connectSignalDisconnected(const Core::Signals::Signal0::slot_type &slot) {
      return disconnectedSignal.connect(slot);
    }
    void disconnectSignalDisconnected(const Core::Signals::Connection &connection) {
      disconnectedSignal.disconnect(connection);
    }
};

}
}

#endif /*MAD_NET_CONNECTION_H_*/