#include "interface/module.h" #include "interface/server.h" #include "interface/event.h" #include "interface/tcpsocket.h" #include "network/include/api.h" #include "core/log.h" //#include //#include #include #include #include // strerror() using interface::Event; namespace network { struct Peer { typedef size_t Id; Id id = 0; sp_ socket; Packet::Type highest_known_type = 99; std::deque socket_buffer; Peer(){} Peer(Id id, sp_ socket): id(id), socket(socket){} }; struct PacketTypeRegistry { sm_ m_types; sm_ m_names; Packet::Type m_next_type = 100; Packet::Type get(const ss_ &name){ auto it = m_types.find(name); if(it != m_types.end()) return it->second; Packet::Type type = m_next_type++; m_types[name] = type; m_names[type] = name; return type; } ss_ get_name(Packet::Type type){ auto it = m_names.find(type); if(it != m_names.end()) return it->second; return ""; } }; struct Module: public interface::Module, public network::Interface { interface::Server *m_server; sp_ m_listening_socket; sm_ m_peers; sm_ m_peers_by_socket; size_t m_next_peer_id = 1; PacketTypeRegistry m_packet_types; Module(interface::Server *server): interface::Module("network"), m_server(server), m_listening_socket(interface::createTCPSocket()) { log_v(MODULE, "network construct"); } ~Module() { log_v(MODULE, "network destruct"); if(m_listening_socket->good()) m_server->remove_socket_event(m_listening_socket->fd()); for(auto pair : m_peers){ const Peer &peer = pair.second; if(peer.socket->good()) m_server->remove_socket_event(peer.socket->fd()); } } void init() { log_v(MODULE, "network init"); m_server->sub_event(this, Event::t("core:start")); m_server->sub_event(this, Event::t("network:listen_event")); m_server->sub_event(this, Event::t("network:incoming_data")); } void event(const Event::Type &type, const Event::Private *p) { EVENT_VOIDN("core:start", on_start) EVENT_TYPEN("network:listen_event", on_listen_event, interface::SocketEvent) EVENT_TYPEN("network:incoming_data", on_incoming_data, interface::SocketEvent) } void on_start() { ss_ address = "any4"; ss_ port = "20000"; if(!m_listening_socket->bind_fd(address, port) || !m_listening_socket->listen_fd()){ log_i(MODULE, "Failed to bind to %s:%s, fd=%i", cs(address), cs(port), m_listening_socket->fd()); } else { log_i(MODULE, "Listening at %s:%s, fd=%i", cs(address), cs(port), m_listening_socket->fd()); } m_server->add_socket_event(m_listening_socket->fd(), Event::t("network:listen_event")); } void on_listen_event(const interface::SocketEvent &event) { log_i(MODULE, "network: on_listen_event(): fd=%i", event.fd); // Create socket sp_ socket(interface::createTCPSocket()); // Accept connection socket->accept_fd(*m_listening_socket.get()); // Store socket Peer::Id peer_id = m_next_peer_id++; m_peers[peer_id] = Peer(peer_id, socket); m_peers_by_socket[socket->fd()] = &m_peers[peer_id]; // Emit event PeerInfo pinfo; pinfo.id = peer_id; pinfo.address = socket->get_remote_address(); m_server->emit_event("network:new_client", new NewClient(pinfo)); m_server->add_socket_event(socket->fd(), Event::t("network:incoming_data")); } void on_incoming_data(const interface::SocketEvent &event) { log_i(MODULE, "network: on_incoming_data(): fd=%i", event.fd); auto it = m_peers_by_socket.find(event.fd); if(it == m_peers_by_socket.end()){ log_w(MODULE, "network: Peer with fd=%i not found", event.fd); return; } Peer &peer = *it->second; int fd = peer.socket->fd(); if(fd != event.fd) throw Exception("on_incoming_data: fds don't match"); char buf[100000]; ssize_t r = recv(fd, buf, 100000, 0); if(r == -1) throw Exception(ss_()+"Receive failed: "+strerror(errno)); if(r == 0){ log_i(MODULE, "Peer %zu disconnected", peer.id); m_server->remove_socket_event(peer.socket->fd()); m_peers_by_socket.erase(peer.socket->fd()); m_peers.erase(peer.id); return; } log_i(MODULE, "Received %zu bytes", r); peer.socket_buffer.insert(peer.socket_buffer.end(), buf, buf + r); for(;;){ if(peer.socket_buffer.size() < 6) return; size_t type = peer.socket_buffer[0]<<0 | peer.socket_buffer[1]<<8; size_t size = peer.socket_buffer[2]<<0 | peer.socket_buffer[3]<<8 | peer.socket_buffer[4]<<16 | peer.socket_buffer[5]<<24; log_i(MODULE, "size=%zu", size); if(peer.socket_buffer.size() < 6 + size) return; log_i(MODULE, "Received full packet; type=%zu, length=6+%zu", type, size); ss_ data(&peer.socket_buffer[6], size); peer.socket_buffer.erase(peer.socket_buffer.begin(), peer.socket_buffer.begin() + 6 + size); // Emit event m_server->emit_event("network:packet_received", new Packet(0, type, data)); } } void send_u(Peer &peer, const Packet::Type &type, const ss_ &data) { // Send new packet types if needed if(m_packet_types.m_next_type > peer.highest_known_type + 1){ Packet::Type highest_known_type_was = peer.highest_known_type; peer.highest_known_type = m_packet_types.m_next_type - 1; for(Packet::Type t1 = highest_known_type_was; t1 < m_packet_types.m_next_type; t1++){ std::ostringstream os(std::ios::binary); os<<(char)((t1>>0) & 0xff); os<<(char)((t1>>8) & 0xff); ss_ name = m_packet_types.get_name(t1); os<<(char)((name.size()>>0) & 0xff); os<<(char)((name.size()>>8) & 0xff); os<<(char)((name.size()>>16) & 0xff); os<<(char)((name.size()>>24) & 0xff); os<>0) & 0xff); os<<(char)((type>>8) & 0xff); os<<(char)((data.size()>>0) & 0xff); os<<(char)((data.size()>>8) & 0xff); os<<(char)((data.size()>>16) & 0xff); os<<(char)((data.size()>>24) & 0xff); os<send_fd(os.str()); } void send_u(PeerInfo::Id recipient, const Packet::Type &type, const ss_ &data) { // Grab Peer (which contains socket) auto it = m_peers.find(recipient); if(it == m_peers.end()){ throw Exception(ss_()+"network::send(): Peer "+itos(recipient) + " doesn't exist"); } Peer &peer = it->second; send_u(peer, type, data); } // Interface Packet::Type packet_type(const ss_ &name) { return m_packet_types.get(name); } void send(PeerInfo::Id recipient, const Packet::Type &type, const ss_ &data) { log_i(MODULE, "network::send()"); send_u(recipient, type, data); } void send(PeerInfo::Id recipient, const ss_ &name, const ss_ &data) { send(recipient, m_packet_types.get(name), data); } void* get_interface() { return dynamic_cast(this); } }; extern "C" { EXPORT void* createModule_network(interface::Server *server){ return (void*)(new Module(server)); } } }