Class: SRT::Server

Inherits:

Object

• Object
• SRT::Server

Defined in:

ext/rbsrt/rbsrt.c

Instance Method Summary

• #close ⇒ Object
• #connection_count ⇒ Object
• #initialize(address, port) ⇒ Object constructor
• #start ⇒ Object

Constructor Details

#initialize(address, port) ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 1454

VALUE rbsrt_server_initialize(VALUE self, VALUE address, VALUE port)
{
    RBSRT_DEBUG_PRINT("server initialize");

    rbsrt_server_t *server;
    int status;
    struct sockaddr_in sa;
    int livemode = SRTT_LIVE;
    int no = 1;

    Check_Type(address, T_STRING);
    Check_Type(port, T_STRING);

    TypedData_Get_Struct(self, rbsrt_server_t, &rbsrt_server_rbtype, server);

    // Initialize struct

    atomic_init(&server->num_connections, 0);


    // Ruby API

    VALUE connections_by_socket = rb_hash_new();

    rb_hash_clear(connections_by_socket);
    
    rb_ivar_set(self, rb_intern("@connections_by_socket"), connections_by_socket);

    
    // create socket

    server->socket = srt_create_socket();

    if (server->socket == SRT_ERROR)
    {
        rb_raise(rb_eStandardError, "%s", srt_getlasterror_str());

        return Qfalse;
    }


    // get address info

    sa.sin_family = AF_INET;
    sa.sin_port = htons(atoi(StringValuePtr(port)));
    
    if (inet_pton(AF_INET, StringValuePtr(address), &sa.sin_addr) != 1)
    {
        rb_raise(rb_eArgError, "bad adddress: %s", strerror(errno));
        
        return Qfalse;
    }
    

    // set up socket

    srt_setsockflag(server->socket, SRTO_TRANSTYPE, &livemode, sizeof(livemode)); // set live mode
    srt_setsockflag(server->socket, SRTO_RCVSYN, &no, sizeof(no)); // non-blocking mode
    srt_setsockflag(server->socket, SRTO_SNDSYN, &no, sizeof(no));
    

    // set up callback

    srt_listen_callback(server->socket, rbsrt_server_listen_callback, (void *)self);

    
    // bind server
    
    status = srt_bind(server->socket, (struct sockaddr *)&sa, sizeof(sa));
    
    if (status == SRT_ERROR)
    {
        rb_raise(rb_eStandardError, "%s", srt_getlasterror_str());
        
        return Qfalse;
    }
    
    
    // listen
    
    status = srt_listen(server->socket, 6); // TODO: Pass backlog size
    
    if (status == SRT_ERROR)
    {
        rb_raise(rb_eStandardError, "%s", srt_getlasterror_str());
        
        return Qfalse;
    }

    return self;
}

Instance Method Details

#close ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 1546

VALUE rbsrt_server_close(VALUE self)
{
    RBSRT_DEBUG_PRINT("server close");

    RBSRT_SERVER_UNWRAP(self, server);

    if (srt_close(server->socket) == SRT_ERROR)
    {
        rbsrt_raise_last_srt_error();
    }

    server->flags &= ~ (RBSRT_SERVER_ACCEPTING | RBSRT_SERVER_BOUND | RBSRT_SERVER_LISTENING);
    server->flags |= RBSRT_SERVER_CLOSED;

    return Qnil;
}

#connection_count ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 1580

VALUE rbsrt_server_connection_count(VALUE self)
{
    RBSRT_SERVER_UNWRAP(self, server);
    
    return SIZET2NUM(RBSRT_SERVER_NUM_CONNECTIONS(server));
}

#start ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 1587

VALUE rbsrt_server_start(VALUE self)
{
    RBSRT_DEBUG_PRINT("server start");

    rb_need_block();

    RBSRT_SERVER_UNWRAP(self, server);

    // rb_iv_set(self, rb_intern("acceptor_block"), rb_block_proc());

    server->acceptor_block = rb_block_proc();

    server->epollid = srt_epoll_create();

    int event_types = SRT_EPOLL_IN;

    srt_epoll_add_usock(server->epollid, server->socket, &event_types);

    SRT_EPOLL_EVENT events[1024];

    struct rbsrt_server_epoll_wait_args args = {
        .server = server,
        .timeout = 100,
        .num_events = 1024,
        .events = events
    };

    struct sockaddr_storage remote_address;
    int addr_size = sizeof(remote_address);
    int remote_fd;
    int running = 1;

    VALUE connections_by_socket = rb_ivar_get(self, rb_intern("@connections_by_socket"));

    while (running)
    {
        int num_events = (int)(uintptr_t)rb_thread_call_without_gvl(rbsrt_server_epoll_wait, &args, RUBY_UBF_IO, 0);

        if (num_events <= 0)
        {
            continue; // timeout
        }

        int read_buf_size = RBSRT_PAYLOAD_SIZE * 8;
        char read_buf[read_buf_size];
        int nbytes;

        // SRT_MSGCTRL msgctl;
        SRT_EPOLL_EVENT *event;
        SRTSOCKET sock;
        SRT_SOCKSTATUS status;
        rbsrt_connection_t *connection;

        VALUE rb_connection;

        int no = 0;
        int no_size = sizeof(no);
        int connection_epoll_events = SRT_EPOLL_OUT | SRT_EPOLL_IN | SRT_EPOLL_ERR;

        for (int i = 0; i < num_events; i++)
        {
            event = &args.events[i];
            sock = event->fd;
            status = srt_getsockstate(sock);
            
            switch (status)
            {
                // accept connections
                case SRTS_LISTENING:

                    remote_fd = srt_accept(sock, (struct sockaddr *)&remote_address, &addr_size);

                    srt_setsockflag(remote_fd, SRTO_RCVSYN, &no, no_size);
                    srt_setsockflag(remote_fd, SRTO_SNDSYN, &no, no_size);

                    connection = malloc(sizeof(rbsrt_connection_t));

                    memset(connection, 0, sizeof(rbsrt_connection_t));

                    rb_connection = TypedData_Make_Struct(mSRTConnectionKlass, rbsrt_connection_t, &rbsrt_connection_rbtype, connection);

                    connection->socket = remote_fd;

                    VALUE should_accept = rb_funcall_with_block(self, rb_intern("instance_exec"), 1, &rb_connection, server->acceptor_block);

                    if (RTEST(should_accept))
                    {
                        srt_epoll_add_usock(server->epollid, connection->socket, &connection_epoll_events);

                        atomic_fetch_add(&server->num_connections, 1);
                        
                        rb_hash_aset(connections_by_socket, INT2FIX(connection->socket), rb_connection);
                    }

                    else
                    {
                        atomic_fetch_sub(&server->num_connections, 1);
                        
                        srt_close(remote_fd);
                    }

                    break;

                case SRTS_CLOSED:
                case SRTS_NONEXIST:
                case SRTS_BROKEN:
                    RBSRT_DEBUG_PRINT("removing connection with socket: %d", sock);

                    srt_epoll_remove_usock(server->epollid, sock);

                    rb_connection = rb_hash_delete(connections_by_socket, INT2FIX(sock));

                    if (RTEST(rb_connection))
                    {
                        if (atomic_fetch_sub(&server->num_connections, 1) == 0)
                        {
                            DEBUG_ERROR_PRINT("removed to many connections");
                        }

                        RBSRT_DEBUG_PRINT("remove connection with socket %d, now %lu sockets", sock, RBSRT_SERVER_NUM_CONNECTIONS(server));

                        RBSRT_CONNECTION_UNWRAP(rb_connection, removed_connection);

                        if (removed_connection->at_close_block)
                        {
                            rb_funcall(removed_connection->at_close_block, rb_intern("call"), 0);

                            removed_connection->at_close_block = 0;
                        }

                        if (removed_connection->at_data_block)
                        {
                            removed_connection->at_data_block = 0;
                        }
                    }

                    break;

                case SRTS_CONNECTED:
                    if (event->events & SRT_EPOLL_IN)
                    {
                        RBSRT_DEBUG_PRINT("will read from socket %d (max bytes %d)", sock, read_buf_size);

                        nbytes = srt_recvmsg2(event->fd, read_buf, read_buf_size, NULL);

                        if (nbytes == SRT_ERROR)
                        {
                            RBSRT_DEBUG_PRINT("failed to read from socket %d: %s", sock, srt_getlasterror_str());

                            break;
                        }

                        if (nbytes > 0)
                        {
                            RBSRT_DEBUG_PRINT("received %d bytes from socket %d", nbytes, sock);

                            rb_connection = rb_hash_lookup(connections_by_socket, INT2FIX(sock));

                            if (RTEST(rb_connection))
                            {
                                RBSRT_DEBUG_PRINT("found connection for socket %d", sock);

                                TypedData_Get_Struct(rb_connection, rbsrt_connection_t, &rbsrt_connection_rbtype, connection);

                                if (connection->at_data_block)
                                {
                                    RBSRT_DEBUG_PRINT("found data handler for socket %d", sock);

                                    VALUE data = rb_str_new(read_buf, nbytes);

                                    rb_funcall(connection->at_data_block, rb_intern("call"), 1, data);
                                }
                            }
                        }
                    }

                    // TODO: Handle other events
                    break;

                default:
                    break;
            }
        }
    }
    
    return Qtrue;
}