v/vlib/x/websocket/websocket_server.v

189 lines
5.1 KiB
V
Raw Normal View History

module websocket
import net
import x.openssl
import log
import sync
import time
import rand
2020-12-04 01:52:26 +01:00
// Server represents a websocket server connection
pub struct Server {
mut:
2020-12-04 01:52:26 +01:00
clients map[string]&ServerClient // clients connected to this server
logger &log.Log // logger used to log
ls net.TcpListener // listener used to get incoming connection to socket
accept_client_callbacks []AcceptClientFn // accept client callback functions
message_callbacks []MessageEventHandler // new message callback functions
close_callbacks []CloseEventHandler // close message callback functions
pub:
2020-12-04 01:52:26 +01:00
port int // port used as listen to incoming connections
is_ssl bool // true if secure connection (not supported yet on server)
pub mut:
ping_interval int = 30
// interval for sending ping to clients (seconds)
2020-12-04 01:52:26 +01:00
state State // current state of connection
}
2020-12-04 01:52:26 +01:00
// ServerClient represents a connected client
struct ServerClient {
pub:
2020-12-04 01:52:26 +01:00
resource_name string // resource that the client access
client_key string // unique key of client
pub mut:
2020-12-28 01:06:57 +01:00
server &Server
client &Client
}
2020-12-04 01:52:26 +01:00
// new_server instance a new websocket server on provided port and route
pub fn new_server(port int, route string) &Server {
return &Server{
port: port
logger: &log.Log{
level: .info
}
state: .closed
}
}
2020-12-02 04:02:53 +01:00
// set_ping_interval sets the interval that the server will send ping messages to clients
pub fn (mut s Server) set_ping_interval(seconds int) {
s.ping_interval = seconds
}
2020-12-04 01:52:26 +01:00
// listen start listen and process to incoming connections from websocket clients
pub fn (mut s Server) listen() ? {
s.logger.info('websocket server: start listen on port $s.port')
2020-12-02 04:02:53 +01:00
s.ls = net.listen_tcp(s.port) ?
s.set_state(.open)
go s.handle_ping()
for {
mut c := s.accept_new_client() or { continue }
go s.serve_client(mut c)
}
s.logger.info('websocket server: end listen on port $s.port')
}
2020-12-04 01:52:26 +01:00
// Close closes server (not implemented yet)
fn (mut s Server) close() {
2020-12-04 01:52:26 +01:00
// TODO: implement close when moving to net from x.net
}
2020-12-02 04:02:53 +01:00
// handle_ping sends ping to all clients every set interval
fn (mut s Server) handle_ping() {
mut clients_to_remove := []string{}
for s.state == .open {
time.sleep(s.ping_interval)
for _, cli in s.clients {
mut c := cli
if c.client.state == .open {
c.client.ping() or {
s.logger.debug('server-> error sending ping to client')
2020-12-02 04:02:53 +01:00
c.client.close(1002, 'Closing connection: ping send error') or {
// we want to continue even if error
continue
}
clients_to_remove << c.client.id
}
if (time.now().unix - c.client.last_pong_ut) > s.ping_interval * 2 {
clients_to_remove << c.client.id
c.client.close(1000, 'no pong received') or { continue }
}
}
}
2020-12-04 01:52:26 +01:00
// TODO: replace for with s.clients.delete_all(clients_to_remove) if (https://github.com/vlang/v/pull/6020) merges
for client in clients_to_remove {
lock {
s.clients.delete(client)
}
}
clients_to_remove.clear()
}
}
2020-12-04 01:52:26 +01:00
// serve_client accepts incoming connection and sets up the callbacks
fn (mut s Server) serve_client(mut c Client) ? {
c.logger.debug('server-> Start serve client ($c.id)')
defer {
c.logger.debug('server-> End serve client ($c.id)')
}
2020-12-02 04:02:53 +01:00
mut handshake_response, mut server_client := s.handle_server_handshake(mut c) ?
accept := s.send_connect_event(mut server_client) ?
if !accept {
s.logger.debug('server-> client not accepted')
2020-12-02 04:02:53 +01:00
c.shutdown_socket() ?
return
}
2020-12-04 01:52:26 +01:00
// the client is accepted
2020-12-02 04:02:53 +01:00
c.socket_write(handshake_response.bytes()) ?
lock {
s.clients[server_client.client.id] = server_client
}
s.setup_callbacks(mut server_client)
c.listen() or {
s.logger.error(err)
return error(err)
}
}
2020-12-02 04:02:53 +01:00
// setup_callbacks initialize all callback functions
fn (mut s Server) setup_callbacks(mut sc ServerClient) {
if s.message_callbacks.len > 0 {
for cb in s.message_callbacks {
if cb.is_ref {
sc.client.on_message_ref(cb.handler2, cb.ref)
} else {
sc.client.on_message(cb.handler)
}
}
}
if s.close_callbacks.len > 0 {
for cb in s.close_callbacks {
if cb.is_ref {
sc.client.on_close_ref(cb.handler2, cb.ref)
} else {
sc.client.on_close(cb.handler)
}
}
}
2020-12-04 01:52:26 +01:00
// set standard close so we can remove client if closed
sc.client.on_close_ref(fn (mut c Client, code int, reason string, mut sc ServerClient) ? {
c.logger.debug('server-> Delete client')
lock {
sc.server.clients.delete(sc.client.id)
}
2020-09-27 10:09:20 +02:00
}, sc)
}
2020-12-04 01:52:26 +01:00
// accept_new_client creates a new client instance for client that connects to the socket
fn (mut s Server) accept_new_client() ?&Client {
2020-12-02 04:02:53 +01:00
mut new_conn := s.ls.accept() ?
c := &Client{
is_server: true
conn: new_conn
ssl_conn: openssl.new_ssl_conn()
logger: s.logger
state: .open
last_pong_ut: time.now().unix
id: rand.uuid_v4()
}
return c
}
// set_state sets current state in a thread safe way
fn (mut s Server) set_state(state State) {
lock {
s.state = state
}
}
2020-12-04 01:52:26 +01:00
// free manages manual free of memory for Server instance
pub fn (mut s Server) free() {
unsafe {
s.clients.free()
s.accept_client_callbacks.free()
s.message_callbacks.free()
s.close_callbacks.free()
}
}