37 #ifndef __PROTOBUF_COMM_SERVER_H_ 38 #define __PROTOBUF_COMM_SERVER_H_ 40 #include <protobuf_comm/frame_header.h> 41 #include <protobuf_comm/message_register.h> 42 #include <protobuf_comm/queue_entry.h> 44 #include <boost/asio.hpp> 45 #include <boost/signals2.hpp> 46 #include <boost/enable_shared_from_this.hpp> 47 #include <google/protobuf/message.h> 49 #ifndef _GLIBCXX_USE_SCHED_YIELD 50 # define _GLIBCXX_USE_SCHED_YIELD 55 #if defined(__GNUC__) && (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 7)) 75 void send(ClientID client, uint16_t component_id, uint16_t msg_type,
76 google::protobuf::Message &m);
77 void send(ClientID client, uint16_t component_id, uint16_t msg_type,
78 std::shared_ptr<google::protobuf::Message> m);
79 void send(ClientID client, std::shared_ptr<google::protobuf::Message> m);
80 void send(ClientID client, google::protobuf::Message &m);
82 void send_to_all(uint16_t component_id, uint16_t msg_type,
83 google::protobuf::Message &m);
84 void send_to_all(uint16_t component_id, uint16_t msg_type,
85 std::shared_ptr<google::protobuf::Message> m);
86 void send_to_all(std::shared_ptr<google::protobuf::Message> m);
95 {
return *message_register_; }
100 boost::signals2::signal<void (ClientID, uint16_t, uint16_t,
101 std::shared_ptr<google::protobuf::Message>)> &
107 boost::signals2::signal<void (ClientID, uint16_t, uint16_t, std::string)> &
113 boost::signals2::signal<void (ClientID, boost::asio::ip::tcp::endpoint &)> &
119 boost::signals2::signal<void (ClientID, const boost::system::error_code &)> &
123 class Session :
public boost::enable_shared_from_this<Session>
127 typedef boost::shared_ptr<Session> Ptr;
130 boost::asio::io_service &io_service);
135 boost::asio::ip::tcp::socket & socket() {
return socket_; }
139 ClientID id()
const {
return id_; }
142 boost::asio::ip::tcp::endpoint & remote_endpoint()
143 {
return remote_endpoint_; }
145 void start_session();
147 void send(uint16_t component_id, uint16_t msg_type,
148 google::protobuf::Message &m);
152 void handle_read_message(
const boost::system::error_code& error);
153 void handle_read_header(
const boost::system::error_code& error);
154 void handle_write(
const boost::system::error_code& error,
160 boost::asio::ip::tcp::socket socket_;
161 boost::asio::ip::tcp::endpoint remote_endpoint_;
164 size_t in_data_size_;
167 std::queue<QueueEntry *> outbound_queue_;
168 std::mutex outbound_mutex_;
169 bool outbound_active_;
175 void handle_accept(Session::Ptr new_session,
const boost::system::error_code& error);
177 void disconnected(boost::shared_ptr<Session> session,
178 const boost::system::error_code &error);
181 boost::asio::io_service io_service_;
182 boost::asio::ip::tcp::acceptor acceptor_;
183 boost::signals2::signal<void (ClientID, uint16_t, uint16_t,
184 std::shared_ptr<google::protobuf::Message>)> sig_rcvd_;
185 boost::signals2::signal<void (ClientID, uint16_t, uint16_t, std::string)> sig_recv_failed_;
186 boost::signals2::signal<void (ClientID, boost::asio::ip::tcp::endpoint &)> sig_connected_;
187 boost::signals2::signal<void (ClientID, const boost::system::error_code &)>
190 std::thread asio_thread_;
192 std::map<ClientID, boost::shared_ptr<Session>> sessions_;
194 #if defined(__GNUC__) && (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 7)) 195 std::atomic<ClientID> next_cid_;
198 std::mutex next_cid_mutex_;
202 bool own_message_register_;
unsigned int ClientID
ID to identify connected clients.
Register to map msg type numbers to Protobuf messages.
boost::signals2::signal< void(ClientID, const boost::system::error_code &)> & signal_disconnected()
Signal that is invoked when a new client has disconnected.
void disconnect(ClientID client)
Disconnect specific client.
void send_to_all(uint16_t component_id, uint16_t msg_type, google::protobuf::Message &m)
Send a message to all clients.
ProtobufStreamServer(unsigned short port)
Constructor.
~ProtobufStreamServer()
Destructor.
void send(ClientID client, uint16_t component_id, uint16_t msg_type, google::protobuf::Message &m)
Send a message to the given client.
Stream server for protobuf message transmission.
boost::signals2::signal< void(ClientID, boost::asio::ip::tcp::endpoint &)> & signal_connected()
Signal that is invoked when a new client has connected.
boost::signals2::signal< void(ClientID, uint16_t, uint16_t, std::string)> & signal_receive_failed()
Signal that is invoked when receiving a message failed.
MessageRegister & message_register()
Get the server's message register.
boost::signals2::signal< void(ClientID, uint16_t, uint16_t, std::shared_ptr< google::protobuf::Message >)> & signal_received()
Signal that is invoked when a message has been received.