Skip to content

GitLab

B
batajelo
Commit ef14f68d authored by louiz’'s avatar louiz’
Browse files
Options

Rename the network::Command class to network::Message 

Makes a lot more sense
parent bf3c7bce
Hide whitespace changes
Inline Side-by-side
Showing with 391 additions and 202 deletions
src/network/client.cpp
View file @ ef14f68d
......@@ -15,9 +15,9 @@ void Client::install_callbacks()
this->install_callback("TRANSFER", std::bind(&Client::transfer_init_callback, this, std::placeholders::_1));
}
void Client::transfer_init_callback(Command* received_command)
void Client::transfer_init_callback(Message* received_message)
{
std::string arg(received_command->body, received_command->body_size);
std::string arg(received_message->body, received_message->body_size);
std::vector<std::string> args;
boost::split(args, arg, boost::is_any_of("|"));
if (args.size() != 3)
......
src/network/client.hpp
View file @ ef14f68d
......@@ -37,7 +37,7 @@ private:
* Called when the server initiate a file transfer with us.
* Install the temporary callback to receive each file chunk.
*/
void transfer_init_callback(Command*);
void transfer_init_callback(Message*);
std::vector<TransferReceiver*> receivers;
boost::asio::io_service io_service;
......
src/network/client_base.cpp
View file @ ef14f68d
#include <logging/logging.hpp>
#include <network/client_base.hpp>
#include <boost/algorithm/string.hpp>
#if defined(_WIN32) || defined(_WIN64)
# include <WinBase.h>
#endif
/**
* Does nothing, it is just used to exit the io_service.run_one() after
......@@ -15,7 +12,7 @@ static void poll_timeout_handler(const boost::system::error_code&)
ClientBase::ClientBase():
BaseIoservice(),
CommandHandler(io_service),
MessageHandler(io_service),
timeout(io_service)
{
}
......@@ -65,13 +62,13 @@ void ClientBase::connect_handler(std::function< void(void) > on_success,
}
}
void ClientBase::ping_callback(Command*)
void ClientBase::ping_callback(Message*)
{
log_debug("Received PING");
Command* command = new Command;
command->set_name("PONG");
this->send(command);
Message* message = new Message;
message->set_name("PONG");
this->send(message);
}
void ClientBase::poll(long timeout)
......
src/network/client_base.hpp
View file @ ef14f68d
......@@ -6,7 +6,7 @@
* The main network class for the clients.
* Connects to the remote server, sends requests to it and
* calls the appropriate callbacks when the answer is received (or
* when a standalone command is received from the server)
* when a standalone message is received from the server)
* @class ClientBase
*/
......@@ -18,13 +18,13 @@
# define __CLIENT_BASE_HPP__
#include <network/base_ioservice.hpp>
#include <network/command_handler.hpp>
#include <network/message_handler.hpp>
#include <network/ping_handler.hpp>
#include <network/timed_event_handler.hpp>
#include <network/command.hpp>
#include <network/message.hpp>
#include <network/timed_event.hpp>
class ClientBase: public BaseIoservice, public CommandHandler,
class ClientBase: public BaseIoservice, public MessageHandler,
public TimedEventHandler, public PingHandler
{
public:
......@@ -59,7 +59,7 @@ private:
/**
* Called when the server sends us a PING request. Sends a PONG back.
*/
void ping_callback(Command*);
void ping_callback(Message*);
boost::asio::deadline_timer timeout;
};
......
src/network/command.cpp → src/network/message.cpp
View file @ ef14f68d
#include <network/command.hpp>
#include <network/message.hpp>
#include <assert.h>
Command::Command():
Message::Message():
body(0),
body_size(0),
callback(nullptr)
{
}
Command::Command(const Command& command)
Message::Message(const Message& message)
{
this->callback = command.callback;
this->header = command.header;
this->name = command.name;
this->set_body(command.body, command.body_size);
this->callback = message.callback;
this->header = message.header;
this->name = message.name;
this->set_body(message.body, message.body_size);
}
Command::~Command()
Message::~Message()
{
delete[] this->body;
}
void Command::set_body(const char* body, int size)
void Message::set_body(const char* body, int size)
{
if (size == -1)
size = ::strlen(body);
......@@ -30,17 +30,17 @@ void Command::set_body(const char* body, int size)
this->set_body_size(size);
}
void Command::set_name(const std::string name)
void Message::set_name(const std::string name)
{
this->name = name;
}
void Command::set_body_size(int size)
void Message::set_body_size(int size)
{
this->body_size = size;
}
void Command::pack()
void Message::pack()
{
assert(this->name.size() > 0);
......
src/network/command.hpp → src/network/message.hpp
View file @ ef14f68d
......@@ -3,23 +3,23 @@
*/
/**
* Represents one single network command.
* A command has a header and a body (also called binary part, which can have
* Represents one single network message, for the base network protocol.
* A message has a header and a body (also called binary part, which can have
* size of 0).
* The header is COMMAND_NAME.BODY_SIZE:
* The header is MESSAGE_NAME.BODY_SIZE:
* The body can be anything, but it is BODY_SIZE bytes long.
* The \n char is not used anywhere in the protocol, and there's no delimiter
* between a body and the header of the next message.
*
* To send a command, a Command object must be created anywhere and filled with
* the correct data and then passed to the CommandHandler::send() method.
* To send a message, a Message object must be created anywhere and filled with
* the correct data and then passed to the MessageHandler::send() method.
* The object will be deleted by the send_handler, after it has been
* succesfully sent.
*
* A Command object is passed by a CommandHandler to the callback associated
* with a command name. This callback is responsible for deleting the command
* A Message object is passed by a MessageHandler to the callback associated
* with a message name. This callback is responsible for deleting the message
* object. (MAYBE)
* @class Command
* @class Message
*/
#include <string>
......@@ -29,15 +29,15 @@
#include <sstream>
#include <functional>
#ifndef __COMMAND_HPP__
# define __COMMAND_HPP__
#ifndef MESSAGE_HPP
# define MESSAGE_HPP
class Command
class Message
{
public:
Command();
Command(const Command&);
~Command();
Message();
Message(const Message&);
~Message();
/**
* Sets the body of the message. A char* will be new[]ed using the size, and
* the data will be copied in it. To avoid that copy, see the body attribute
......@@ -51,7 +51,7 @@ public:
*/
void set_body_size(int size);
/**
* This must be called before the object is passed to CommandHandler::send(),
* This must be called before the object is passed to MessageHandler::send(),
* it will set the header correctly, using the body size etc.
*/
void pack();
......@@ -72,7 +72,7 @@ public:
std::function< void(void) > callback;
private:
Command& operator=(const Command&);
Message& operator=(const Message&);
};
#endif // __COMMAND_HPP__
#endif // __MESSAGE_HPP__
src/network/command_handler.cpp → src/network/message_handler.cpp
View file @ ef14f68d
#include <logging/logging.hpp>
#include <network/command_handler.hpp>
#include <network/command.hpp>
#include <network/message_handler.hpp>
#include <network/message.hpp>
CommandHandler::CommandHandler(boost::asio::io_service& io_service):
MessageHandler::MessageHandler(boost::asio::io_service& io_service):
BaseSocket(io_service),
writing(false)
{
}
CommandHandler::~CommandHandler()
MessageHandler::~MessageHandler()
{
}
void CommandHandler::install_callback(const std::string& command,
t_read_callback callback)
void MessageHandler::install_callback(const std::string& message,
t_read_callback callback)
{
log_debug("installing callback for command " << command);
this->callbacks[command] = callback;
log_debug("installing callback for message " << message);
this->callbacks[message] = callback;
}
void CommandHandler::install_callback_once(const std::string& command,
t_read_callback callback)
void MessageHandler::install_callback_once(const std::string& message,
t_read_callback callback)
{
log_debug("installing ONCE callback for command " << command);
this->callbacks_once[command] = callback;
log_debug("installing ONCE callback for message " << message);
this->callbacks_once[message] = callback;
}
void CommandHandler::remove_callback(const std::string& command)
void MessageHandler::remove_callback(const std::string& message)
{
std::map<const std::string, t_read_callback >::iterator it;
it = this->callbacks.find(command);
auto it = this->callbacks.find(message);
if (it != this->callbacks.end())
this->callbacks.erase(it);
else
log_warning("Could not remove callback: " << command);
log_warning("Could not remove callback: " << message);
}
t_read_callback CommandHandler::get_callback(const std::string& command)
t_read_callback MessageHandler::get_callback(const std::string& message)
{
std::map<const std::string, t_read_callback >::iterator it;
it = this->callbacks.find(command);
auto it = this->callbacks.find(message);
if (it != this->callbacks.end())
return it->second;
it = this->callbacks_once.find(command);
it = this->callbacks_once.find(message);
if (it != this->callbacks_once.end())
{
log_debug("Removing callback for command " << command);
log_debug("Removing callback for message " << message);
t_read_callback callback = it->second;
this->callbacks_once.erase(it);
return callback;
......@@ -56,7 +52,7 @@ t_read_callback CommandHandler::get_callback(const std::string& command)
return 0;
}
void CommandHandler::read_handler(const boost::system::error_code& error, const std::size_t bytes_transferred)
void MessageHandler::read_handler(const boost::system::error_code& error, const std::size_t bytes_transferred)
{
log_debug("read_handler, size: " << bytes_transferred << " bytes.");
if (error)
......@@ -67,7 +63,7 @@ void CommandHandler::read_handler(const boost::system::error_code& error, const
}
if (bytes_transferred <= 1)
{
log_warning("Not enough data received for a command header to be valid.");
log_warning("Not enough data received for a message header to be valid.");
return ;
}
......@@ -81,40 +77,40 @@ void CommandHandler::read_handler(const boost::system::error_code& error, const
while (c[pos] && c[pos] != '.')
pos++;
std::string command_name;
std::string message_name;
std::size_t size;
if (pos == bytes_transferred)
{ // no '.' was found, the command name is assumed to 1 char long.
command_name = std::string(1, c[0]);
{ // no '.' was found, the message name is assumed to 1 char long.
message_name = std::string(1, c[0]);
size = atoi(std::string(c+1, bytes_transferred-2).data());;
}
else
{
command_name = std::string(c, pos);
message_name = std::string(c, pos);
size = atoi(std::string(c+pos+1, bytes_transferred-pos-2).data()); // remove the ending :
}
log_debug("Received : " << command_name << " . " << size);
Command* command = new Command;
command->set_name(command_name);
log_debug("Received : " << message_name << " . " << size);
Message* message = new Message;
message->set_name(message_name);
delete[] c;
// Find out if a callback was registered for that command.
t_read_callback callback = this->get_callback(command_name);
// Find out if a callback was registered for that message.
t_read_callback callback = this->get_callback(message_name);
// We check what we need to read on the socket to have the rest of the binary datas
const std::size_t length_to_read = this->data.size() >= size ? 0 : size - this->data.size();
boost::asio::async_read(this->socket,
this->data,
boost::asio::transfer_at_least(length_to_read),
std::bind(&CommandHandler::binary_read_handler, this,
std::bind(&MessageHandler::binary_read_handler, this,
std::placeholders::_1,
command,
message,
size, callback));
}
void CommandHandler::binary_read_handler(const boost::system::error_code& error,
Command* command,
void MessageHandler::binary_read_handler(const boost::system::error_code& error,
Message* message,
std::size_t bytes_transferred,
t_read_callback callback)
{
......@@ -124,84 +120,84 @@ void CommandHandler::binary_read_handler(const boost::system::error_code& error,
exit(1);
}
log_debug("binary_read_handler " << bytes_transferred);
command->body = new char[bytes_transferred];
this->data.sgetn(command->body, bytes_transferred);
command->set_body_size(bytes_transferred);
message->body = new char[bytes_transferred];
this->data.sgetn(message->body, bytes_transferred);
message->set_body_size(bytes_transferred);
if (callback)
callback(command);
callback(message);
else
log_debug("no callback");
delete command;
delete message;
this->install_read_handler();
}
void CommandHandler::install_read_handler(void)
void MessageHandler::install_read_handler(void)
{
boost::asio::async_read_until(this->socket,
this->data,
':',
std::bind(&CommandHandler::read_handler, this,
std::bind(&MessageHandler::read_handler, this,
std::placeholders::_1,
std::placeholders::_2));
}
void CommandHandler::request_answer(Command* command, t_read_callback on_answer, std::string name)
void MessageHandler::request_answer(Message* message, t_read_callback on_answer, std::string name)
{
if (name.size() == 0)
name = command->get_name();
name = message->get_name();
this->install_callback_once(name, on_answer);
this->send(command);
this->send(message);
}
void CommandHandler::send(Command* command, std::function< void(void) > on_sent)
void MessageHandler::send(Message* message, std::function< void(void) > on_sent)
{
if (on_sent)
command->callback = on_sent;
log_debug("Sending command: " << command->get_name());
this->commands_to_send.push_front(command);
this->check_commands_to_send();
message->callback = on_sent;
log_debug("Sending message: " << message->get_name());
this->messages_to_send.push_front(message);
this->check_messages_to_send();
}
bool CommandHandler::check_commands_to_send()
bool MessageHandler::check_messages_to_send()
{
log_debug("Length of the queue: " << this->commands_to_send.size());
if (this->writing || this->commands_to_send.empty())
log_debug("Length of the queue: " << this->messages_to_send.size());
if (this->writing || this->messages_to_send.empty())
return false;
this->actually_send(this->commands_to_send.back());
this->commands_to_send.pop_back();
this->actually_send(this->messages_to_send.back());
this->messages_to_send.pop_back();
return true;
}
void CommandHandler::actually_send(Command* command)
void MessageHandler::actually_send(Message* message)
{
this->writing = true;
command->pack();
message->pack();
std::vector<boost::asio::const_buffer> buffs;
buffs.push_back(boost::asio::buffer(command->header.data(), command->header.length()));
buffs.push_back(boost::asio::buffer(command->body, command->body_size));
buffs.push_back(boost::asio::buffer(message->header.data(), message->header.length()));
buffs.push_back(boost::asio::buffer(message->body, message->body_size));
async_write(this->socket,
buffs,
std::bind(&CommandHandler::send_handler, this,
std::bind(&MessageHandler::send_handler, this,
std::placeholders::_1,
std::placeholders::_2,
command));
message));
}
void CommandHandler::send_handler(const boost::system::error_code& error,
void MessageHandler::send_handler(const boost::system::error_code& error,
std::size_t bytes_transferred,
Command* command)
Message* message)
{
this->writing = false;
assert(bytes_transferred == command->header.length() + command->body_size);
assert(bytes_transferred == message->header.length() + message->body_size);
if (command->callback)
command->callback();
delete command;
if (message->callback)
message->callback();
delete message;
// TODO check for error
if (error)
exit(1);
this->check_commands_to_send();
this->check_messages_to_send();
}
src/network/command_handler.hpp → src/network/message_handler.hpp
View file @ ef14f68d
......@@ -4,13 +4,13 @@
/**
* Handles the reading part of the socket.
* Keeps a list of callbacks to call, associated with a command name, and
* execute them when the associated command is received.
* @class CommandHandler
* Keeps a list of callbacks to call, associated with a message name, and
* execute them when the associated message is received.
* @class MessageHandler
*/
#ifndef __COMMAND_HANDLER_HPP__
# define __COMMAND_HANDLER_HPP__
#ifndef MESSAGE_HANDLER_HPP
# define MESSAGE_HANDLER_HPP
#include <deque>
#include <map>
......@@ -20,44 +20,43 @@
#include <network/transfer_sender.hpp>
#include <network/base_socket.hpp>
#include <network/command.hpp>
#include <network/message.hpp>
typedef std::function<void(Command*)> t_read_callback;
typedef std::deque<Command*> command_queue;
typedef std::function<void(Message*)> t_read_callback;
class CommandHandler: public BaseSocket
class MessageHandler: public BaseSocket
{
friend void TransferSender::send_next_chunk();
public:
explicit CommandHandler(boost::asio::io_service&);
virtual ~CommandHandler();
explicit MessageHandler(boost::asio::io_service&);
virtual ~MessageHandler();
void install_read_handler();
/**
* called when there's something to read on the socket. Reads the command
* called when there's something to read on the socket. Reads the message
* the size of the arguments, and then calls binary_read_handler to read
* the arguments of the commad, if any.
* the arguments of the message, if any.
*/
void read_handler(const boost::system::error_code& error, const std::size_t bytes_transferred);
/**
* Read the arguments after a command (can read 0 bytes too) and pass that
* to the callback that was associated with this command.
* Read the arguments after a message (can read 0 bytes too) and pass that
* to the callback that was associated with this message.
*/
void binary_read_handler(const boost::system::error_code&, Command*, std::size_t, t_read_callback);
void binary_read_handler(const boost::system::error_code&, Message*, std::size_t, t_read_callback);
/**
* Sends a command, and use install_callback_once to wait for the answer
* Sends a message, and use install_callback_once to wait for the answer
* and call that callback to handle it.
*/
void request_answer(Command*, t_read_callback on_answer, std::string name = "");
void request_answer(Message*, t_read_callback on_answer, std::string name = "");
/**
* Install a new callback associated with a command. This callback will
* be called upon receiving that command.
* Install a new callback associated with a message. This callback will
* be called upon receiving that message.
*/
void install_callback(const std::string&, t_read_callback);
/**
* Install a new callback associated with a command. This callback will
* be called upon receiving that command, but only once. This is used
* for example if you send a command waiting for and answer, you install
* Install a new callback associated with a message. This callback will
* be called upon receiving that message, but only once. This is used
* for example if you send a message waiting for and answer, you install
* a callback that will handle that answer, and only this one.
*/
void install_callback_once(const std::string&, t_read_callback);
......@@ -66,16 +65,16 @@ public:
*/
void remove_callback(const std::string&);
/**
* Add the given command to the commands_to_send queue, then
* calls check_commands_to_send. Which may send the next available
* command, if there's no async_write() call already running.
* It does not necessarily actually send the command on the socket.
* Add the given message to the messages_to_send queue, then
* calls check_messages_to_send. Which may send the next available
* message, if there's no async_write() call already running.
* It does not necessarily actually send the message on the socket.
*/
void send(Command* command, std::function< void(void) > on_sent = 0);
void send(Message* message, std::function< void(void) > on_sent = 0);
protected:
/**
* Returns the callback associated with the passed command name.
* Returns the callback associated with the passed message name.
* Returns 0 if nothing was found, in that case the execution of the
* return value cause a failure.
*/
......@@ -83,18 +82,18 @@ protected:
/**
* @todo Check if the data was correctly sent on the socket
*/
void send_handler(const boost::system::error_code&, std::size_t, Command*);
void send_handler(const boost::system::error_code&, std::size_t, Message*);
/**
* Actually sends the command on the socket, calling async_write.
* Actually sends the message on the socket, calling async_write.
*/
void actually_send(Command*);
void actually_send(Message*);
/**
* Checks if there's something to send on the socket. We first check if
* the writing boolean is false, and then we pop the next available command
* the writing boolean is false, and then we pop the next available message
* from the queue (if any) and we send it using async_write.
* @returns true if a command was sent, false otherwise.
* @returns true if a message was sent, false otherwise.
*/
bool check_commands_to_send();
bool check_messages_to_send();
/**
* A buffer keeping the data that is read on the socket.
*/
......@@ -105,8 +104,8 @@ protected:
virtual void on_connection_closed() = 0;
private:
CommandHandler(const CommandHandler&);
CommandHandler& operator=(const CommandHandler&);
MessageHandler(const MessageHandler&);
MessageHandler& operator=(const MessageHandler&);
std::map<const std::string, t_read_callback > callbacks;
std::map<const std::string, t_read_callback > callbacks_once;
......@@ -114,14 +113,13 @@ private:
* A queue of messages. If there's not async_write running, we pop one
* from it and we send it.