path: root/src/common/network.c

//
// Network Subsystem (previously known as socket system)
//
// Author: Florian Wilkemeyer <fw@f-ws.de>
//
// Copyright (c) rAthena Project (www.rathena.org) - Licensed under GNU GPL
// For more information, see LICENCE in the main folder
//
//
//#ifdef HAVE_ACCETP4
#define _GNU_SOURCE
//#endif

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/fcntl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>


#include "../common/cbasetypes.h"
#include "../common/showmsg.h"
#include "../common/timer.h"
#include "../common/evdp.h"
#include "../common/netbuffer.h"

#include "../common/network.h"

#define ENABLE_IPV6
#define HAVE_ACCEPT4
#define EVENTS_PER_CYCLE 10
#define PARANOID_CHECKS

// Local Vars (settings..)
static int l_ListenBacklog = 64;

//
// Global Session Array (previously exported as session[]
//
SESSION g_Session[MAXCONN];


//
static bool onSend(int32 fd);


#define _network_free_netbuf_async( buf ) add_timer( 0, _network_async_free_netbuf_proc, 0,  (intptr_t) buf)
static int _network_async_free_netbuf_proc(int tid, unsigned int tick, int id, intptr_t data)
{
    // netbuf is in data
    netbuffer_put((netbuf)data);

    return 0;
}//end: _network_async_free_netbuf_proc()



void network_init()
{
    SESSION *s;
    int32 i;

    memset(g_Session, 0x00, (sizeof(SESSION) * MAXCONN));

    for (i = 0; i < MAXCONN; i++) {
        s = &g_Session[i];

        s->type = NST_FREE;
        s->disconnect_in_progress = false;

    }

    // Initialize the correspondig event dispatcher
    evdp_init();

    //
    add_timer_func_list(_network_async_free_netbuf_proc, "_network_async_free_netbuf_proc");

}//end: network_init()


void network_final()
{

    // @TODO:
    // .. disconnect and cleanup everything!

    evdp_final();

}//end: network_final()


void network_do()
{
    struct EVDP_EVENT l_events[EVENTS_PER_CYCLE];
    register struct EVDP_EVENT *ev;
    register int n, nfds;
    register SESSION *s;

    nfds = evdp_wait(l_events,  EVENTS_PER_CYCLE, 1000);  // @TODO: timer_getnext()

    for (n = 0; n < nfds; n++) {
        ev = &l_events[n];
        s = &g_Session[ ev->fd ];

        if (ev->events & EVDP_EVENT_HUP) {
            network_disconnect(ev->fd);
            continue; // no further event processing.
        }// endif vent is HUP (disconnect)


        if (ev->events & EVDP_EVENT_IN) {

            if (s->onRecv != NULL) {
                if (false == s->onRecv(ev->fd)) {
                    network_disconnect(ev->fd);
                    continue; // ..
                }
            } else {
                ShowError("network_do: fd #%u has no onRecv proc set. - disconnecting\n", ev->fd);
                network_disconnect(ev->fd);
                continue;
            }

        }// endif event is IN (recv)


        if (ev->events & EVDP_EVENT_OUT) {
            if (s->onSend != NULL) {
                if (false == s->onSend(ev->fd)) {
                    network_disconnect(ev->fd);
                    continue;
                }
            } else {
                ShowError("network_do: fd #%u has no onSend proc set. - disconnecting\n", ev->fd);
                network_disconnect(ev->fd);
                continue;
            }
        }// endif event is OUT (send)

    }//endfor

}//end: network_do()


static bool _setnonblock(int32 fd)
{
    int flags = fcntl(fd, F_GETFL, 0);
    if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
        return false;

    return true;
}//end: _setnonblock()


static bool _network_accept(int32 fd)
{
    SESSION *listener = &g_Session[fd];
    SESSION *s;
    union {
        struct sockaddr_in v4;
#ifdef ENABLE_IPV6
        struct sockaddr_in6 v6;
#endif
    } _addr;
    int newfd;
    socklen_t addrlen;
    struct sockaddr *addr;

    // Accept until OS returns - nothing to accept anymore
    // - this is required due to our EVDP abstraction. (which handles on listening sockets similar to epoll's EPOLLET flag.)
    while (1) {
#ifdef ENABLE_IPV6
        if (listener->v6 == true) {
            addrlen = sizeof(_addr.v6);
            addr = (struct sockaddr *)&_addr.v6;
        } else {
#endif
            addrlen = sizeof(_addr.v4);
            addr = (struct sockaddr *)&_addr.v4;
#ifdef ENABLE_IPV6
        }
#endif

#ifdef HAVE_ACCEPT4
        newfd = accept4(fd, addr, &addrlen, SOCK_NONBLOCK);
#else
        newfd = accept(fd, addr, &addrlen);
#endif

        if (newfd == -1) {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                break; // this is fully valid & whished., se explaination on top of while(1)

            // Otherwis .. we have serious problems :( seems tahat our listner has gone away..
            // @TODO handle this ..
            ShowError("_network_accept: accept() returned error. closing listener. (errno: %u / %s)\n", errno, strerror(errno));

            return false; // will call disconnect after return.
            //break;
        }

#ifndef HAVE_ACCEPT4 // no accept4 means, we have to set nonblock by ourself. ..
        if (_setnonblock(newfd) == false) {
            ShowError("_network_accept: failed to set newly accepted connection nonblocking (errno: %u / %s). - disconnecting.\n", errno, strerror(errno));
            close(newfd);
            continue;
        }
#endif

        // Check connection limits.
        if (newfd >= MAXCONN) {
            ShowError("_network_accept: failed to accept connection - MAXCONN (%u) exceeded.\n", MAXCONN);
            close(newfd);
            continue; // we have to loop over the events (and disconnect them too ..) but otherwise we would leak event notifications.
        }


        // Create new Session.
        s = &g_Session[newfd];
        s->type = NST_CLIENT;

        // The new connection inherits listenr's handlers.
        s->onDisconnect = listener->onDisconnect;
        s->onConnect = listener->onConnect; // maybe useless but .. fear the future .. :~

        // Register the new connection @ EVDP
        if (evdp_addclient(newfd, &s->evdp_data) == false) {
            ShowError("_network_accept: failed to accept connection - event subsystem returned an error.\n");
            close(newfd);
            s->type = NST_FREE;
        }

        // Call the onConnect handler on the listener.
        if (listener->onConnect(newfd) == false) {
            // Resfused by onConnect handler..
            evdp_remove(newfd, &s->evdp_data);

            close(newfd);
            s->type = NST_FREE;

            s->data = NULL; // be on the safe side ~ !
            continue;
        }


    }

    return true;
}//end: _network_accept()


void network_disconnect(int32 fd)
{
    SESSION *s = &g_Session[fd];
    netbuf b, bn;

    // Prevent recursive calls
    // by wrong implemented on disconnect handlers.. and such..
    if (s->disconnect_in_progress == true)
        return;

    s->disconnect_in_progress = true;


    // Disconnect Todo:
    //  - Call onDisconnect Handler
    //  - Release all Assigned buffers.
    //  - remove from event system (notifications)
    //  - cleanup session structure
    //  - close connection.
    //

    if (s->onDisconnect != NULL &&
        s->type != NST_LISTENER) {

        s->onDisconnect(fd);
    }

    // Read Buffer
    if (s->read.buf != NULL) {
        netbuffer_put(s->read.buf);
        s->read.buf = NULL;
    }

    // Write Buffer(s)
    b = s->write.buf;
    while (1) {
        if (b == NULL) break;

        bn = b->next;

        netbuffer_put(b);

        b = bn;
    }
    s->write.buf = NULL;
    s->write.buf_last = NULL;

    s->write.n_outstanding = 0;
    s->write.max_outstanding = 0;


    // Remove from event system.
    evdp_remove(fd, &s->evdp_data);

    // Cleanup Session Structure.
    s->type = NST_FREE;
    s->data = NULL; // no application level data assigned
    s->disconnect_in_progress = false;


    // Close connection
    close(fd);

}//end: network_disconnect()


int32 network_addlistener(bool v6,  const char *addr,  uint16 port)
{
    SESSION *s;
    int optval, fd;

#if !defined(ENABLE_IPV6)
    if (v6 == true) {
        ShowError("network_addlistener(%c, '%s', %u):  this release has no IPV6 support.\n", (v6==true?'t':'f'),  addr, port);
        return -1;
    }
#endif


#ifdef ENABLE_IPV6
    if (v6 == true)
        fd = socket(AF_INET6, SOCK_STREAM, 0);
    else
#endif
        fd = socket(AF_INET, SOCK_STREAM, 0);

    // Error?
    if (fd == -1) {
        ShowError("network_addlistener(%c, '%s', %u):  socket() failed (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port,    errno, strerror(errno));
        return -1;
    }

    // Too many connections?
    if (fd >= MAXCONN) {
        ShowError("network_addlistener(%c, '%s', %u):  cannot create listener, exceeds more than supported connections (%u).\n", (v6==true?'t':'f'),  addr, port, MAXCONN);
        close(fd);
        return -1;
    }


    s = &g_Session[fd];
    if (s->type != NST_FREE) { // additional checks.. :)
        ShowError("network_addlistener(%c, '%s', %u): failed, got fd #%u which is already in use in local session table?!\n", (v6==true?'t':'f'),  addr, port, fd);
        close(fd);
        return -1;
    }


    // Fill ip addr structs
#ifdef ENABLE_IPV6
    if (v6 == true) {
        memset(&s->addr.v6, 0x00, sizeof(s->addr.v6));
        s->addr.v6.sin6_family = AF_INET6;
        s->addr.v6.sin6_port = htons(port);
        if (inet_pton(AF_INET6, addr, &s->addr.v6.sin6_addr) != 1) {
            ShowError("network_addlistener(%c, '%s', %u): failed to parse the given IPV6 address.\n", (v6==true?'t':'f'),  addr, port);
            close(fd);
            return -1;
        }

    } else {
#endif
        memset(&s->addr.v4, 0x00, sizeof(s->addr.v4));
        s->addr.v4.sin_family = AF_INET;
        s->addr.v4.sin_port = htons(port);
        s->addr.v4.sin_addr.s_addr = inet_addr(addr);
#ifdef ENABLE_IPV6
    }
#endif


    // if OS has support for SO_REUSEADDR, apply the flag
    // so the address could be used when there're still time_wait sockets outstanding from previous application run.
#ifdef SO_REUSEADDR
    optval=1;
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
#endif

    // Bind
#ifdef ENABLE_IPV6
    if (v6 == true) {
        if (bind(fd, (struct sockaddr *)&s->addr.v6,  sizeof(s->addr.v6)) == -1) {
            ShowError("network_addlistener(%c, '%s', %u): bind failed (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));
            close(fd);
            return -1;
        }
    } else {
#endif
        if (bind(fd, (struct sockaddr *)&s->addr.v4,  sizeof(s->addr.v4)) == -1) {
            ShowError("network_addlistener(%c, '%s', %u): bind failed (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));
            close(fd);
            return -1;
        }
#ifdef ENABLE_IPV6
    }
#endif

    if (listen(fd, l_ListenBacklog) == -1) {
        ShowError("network_addlistener(%c, '%s', %u): listen failed (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));
        close(fd);
        return -1;
    }


    // Set to nonblock!
    if (_setnonblock(fd) == false) {
        ShowError("network_addlistener(%c, '%s', %u): cannot set to nonblock (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));
        close(fd);
        return -1;
    }


    // Rgister @ evdp.
    if (evdp_addlistener(fd, &s->evdp_data) != true) {
        ShowError("network_addlistener(%c, '%s', %u): eventdispatcher subsystem returned an error.\n", (v6==true?'t':'f'),  addr, port);
        close(fd);
        return -1;
    }


    // Apply flags on Session array for this conneciton.
    if (v6 == true)  s->v6 = true;
    else            s->v6 = false;

    s->type = NST_LISTENER;
    s->onRecv = _network_accept;

    ShowStatus("Added Listener on '%s':%u\n", addr, port, (v6==true ? "(ipv6)":"(ipv4)"));

    return fd;
}//end: network_addlistener()


static bool _network_connect_establishedHandler(int32 fd)
{
    register SESSION *s = &g_Session[fd];
    int val;
    socklen_t val_len;

    if (s->type == NST_FREE)
        return true;    // due to multiple non coalesced event notifications
    // this can happen .. when a previous handled event has already disconnected the connection
    // within the same cycle..

    val = -1;
    val_len = sizeof(val);
    getsockopt(fd, SOL_SOCKET, SO_ERROR, &val, &val_len);

    if (val != 0) {
        // :( .. cleanup session..
        s->type = NST_FREE;
        s->onSend = NULL;
        s->onConnect = NULL;
        s->onDisconnect = NULL;

        evdp_remove(fd, &s->evdp_data);
        close(fd);

        return true; // we CANT return false,
        // becuase the normal disconnect procedure would execute the ondisconnect handler, which we dont want .. in this case.
    } else {
        // ok
        if (s->onConnect(fd) == false) {
            // onConnect handler has refused the connection ..
            // cleanup .. and ok
            s->type = NST_FREE;
            s->onSend = NULL;
            s->onConnect = NULL;
            s->onDisconnect = NULL;

            evdp_remove(fd, &s->evdp_data);
            close(fd);

            return true; // we dnot want the ondisconnect handler to be executed, so its okay to handle this by ourself.
        }

        // connection established !
        //
        if (evdp_outgoingconnection_established(fd, &s->evdp_data) == false) {
            return false; // we want the normal disconnect procedure.. with call to ondisconnect handler.
        }

        s->onSend = NULL;

        ShowStatus("#%u connection successfull!\n", fd);
    }

    return true;
}//end: _network_connect_establishedHandler()


int32 network_connect(bool v6,
                      const char *addr,
                      uint16 port,
                      const char *from_addr,
                      uint16 from_port,
                      bool (*onConnectionEstablishedHandler)(int32 fd),
                      void (*onConnectionLooseHandler)(int32 fd)
                     )
{
    register SESSION *s;
    int32 fd, optval, ret;
    struct sockaddr_in ip4;
#ifdef ENABLE_IPV6
    struct sockaddr_in6 ip6;
#endif

#ifdef ENABLE_IPV6
    if (v6 == true)
        fd = socket(AF_INET6, SOCK_STREAM, 0);
    else
#endif
        fd = socket(AF_INET, SOCK_STREAM, 0);

#ifndef ENABLE_IPV6
    // check..
    if (v6 == true) {
        ShowError("network_connect(%c, '%s', %u...): tried to create an ipv6 connection, IPV6 is not supported in this release.\n", (v6==true?'t':'f'),  addr, port);
        return -1;
    }
#endif

    // check connection limits.
    if (fd >= MAXCONN) {
        ShowError("network_connect(%c, '%s', %u...): cannot create new connection, exceeeds more than supported connections (%u)\n", (v6==true?'t':'f'),  addr, port);
        close(fd);
        return -1;
    }


    // Originating IP/Port pair given ?
    if (from_addr != NULL && *from_addr != 0) {
        //..
#ifdef SO_REUSEADDR
        optval=1;
        setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
#endif

#ifdef ENABLE_IPV6
        if (v6 == true) {
            memset(&ip6, 0x00, sizeof(ip6));
            ip6.sin6_family = AF_INET6;
            ip6.sin6_port = htons(from_port);

            if (inet_pton(AF_INET6, from_addr, &ip6.sin6_addr) != 1) {
                ShowError("network_connect(%c, '%s', %u...): cannot parse originating (from) IPV6 address (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));
                close(fd);
                return -1;
            }

            ret = bind(fd, (struct sockaddr *)&ip6, sizeof(ip6));
        } else {
#endif
            memset(&ip4, 0x00, sizeof(ip4));

            ip4.sin_family = AF_INET;
            ip4.sin_port = htons(from_port);
            ip4.sin_addr.s_addr = inet_addr(from_addr);
            ret = bind(fd, (struct sockaddr *)&ip4, sizeof(ip4));
#ifdef ENABLE_IPV6
        }
#endif

    }


    // Set non block
    if (_setnonblock(fd) == false) {
        ShowError("network_connect(%c, '%s', %u...): cannot set socket to nonblocking (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));
        close(fd);
        return -1;
    }


    // Create ip addr block to connect to ..
#ifdef ENABLE_IPV6
    if (v6 == true) {
        memset(&ip6, 0x00, sizeof(ip6));
        ip6.sin6_family = AF_INET6;
        ip6.sin6_port = htons(port);

        if (inet_pton(AF_INET6, addr, &ip6.sin6_addr) != 1) {
            ShowError("network_connect(%c, '%s', %u...): cannot parse destination IPV6 address (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));
            close(fd);
            return -1;
        }

    } else {
#endif
        memset(&ip4, 0x00, sizeof(ip4));

        ip4.sin_family = AF_INET;
        ip4.sin_port = htons(port);
        ip4.sin_addr.s_addr = inet_addr(addr);
#ifdef ENABLE_IPV6
    }
#endif


    // Assign Session..
    s = &g_Session[fd];
    s->type = NST_OUTGOING;
    s->v6 = v6;
    s->onConnect = onConnectionEstablishedHandler;
    s->onDisconnect = onConnectionLooseHandler;
    s->onRecv = NULL;
    s->onSend = _network_connect_establishedHandler;
#ifdef ENABLE_IPV6
    if (v6 == true)
        memcpy(&s->addr.v6, &ip6, sizeof(ip6));
    else
#endif
        memcpy(&s->addr.v4, &ip4, sizeof(ip4));

    // Register @ EVDP. as outgoing (see doc of the function)
    if (evdp_addconnecting(fd, &s->evdp_data) == false) {
        ShowError("network_connect(%c, '%s', %u...): eventdispatcher subsystem returned an error.\n", (v6==true?'t':'f'),  addr, port);

        // cleanup session x.x..
        s->type = NST_FREE;
        s->onConnect = NULL;
        s->onDisconnect = NULL;
        s->onSend = NULL;

        // close, return error code.
        close(fd);
        return -1;
    }


#ifdef ENABLE_IPV6
    if (v6 == true)
        ret = connect(fd, (struct sockaddr *)&ip6, sizeof(ip6));
    else
#endif
        ret = connect(fd, (struct sockaddr *)&ip4, sizeof(ip4));


    //
    if (ret != 0 && errno != EINPROGRESS) {
        ShowWarning("network_connect(%c, '%s', %u...): connection failed (errno: %u / %s)\n", (v6==true?'t':'f'),  addr, port, errno, strerror(errno));

        // Cleanup session ..
        s->type = NST_FREE;
        s->onConnect = NULL;
        s->onDisconnect = NULL;
        s->onSend = NULL;

        // .. remove from evdp and close fd.
        evdp_remove(fd, &s->evdp_data);
        close(fd);
        return -1;
    }


    // ! The Info Message :~D
    ShowStatus("network_connect fd#%u (%s:%u) in progress.. \n", fd, addr, port);

    return fd;
}//end: network_connect()


static bool _onSend(int32 fd)
{
    register SESSION *s = &g_Session[fd];
    register netbuf buf, buf_next;
    register uint32 szNeeded;
    register int wLen;

    if (s->type == NST_FREE)
        return true;    // Possible due to multipl non coalsced event notifications
    // so onSend gets called after disconnect caused by an previous vent.
    // we can ignore the call to onSend, then.

    buf = s->write.buf;
    while (1) {
        if (buf == NULL)
            break;

        buf_next = buf->next;


        szNeeded = (buf->dataLen - s->write.dataPos);   // using th session-local .dataPos member, due to shared write buffer support.

        // try to write.
        wLen = write(fd, &buf->buf[s->write.dataPos],  szNeeded);
        if (wLen == 0) {
            return false; // eof.
        } else if (wLen == -1) {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                return true; // dont disconnect / try again later.

            // all other errors. .
            return false;
        }

        // Wrote data.. =>
        szNeeded -= wLen;
        if (szNeeded > 0) {
            // still data left ..
            //
            s->write.dataPos += wLen; // fix offset.
            return true;
        } else {
            // this buffer has been written successfully
            // could be returned to pool.
            netbuffer_put(buf);
            s->write.n_outstanding--; // When threadsafe -> Interlocked here.
            s->write.dataPos = 0;
        }


        buf = buf_next;
    }

    // okay,
    // reaching this part means:
    // while interrupted by break -
    // which means all buffers are written, nothing left
    //

    s->write.buf_last = NULL;
    s->write.buf = NULL;
    s->write.n_outstanding = 0;
    s->write.dataPos = 0;

    // Remove from event dispatcher (write notification)
    //
    evdp_writable_remove(fd, &s->evdp_data);

    return true;
}//end: _onSend()


static bool _onRORecv(int32 fd)
{
    register SESSION *s = &g_Session[fd];
    register uint32 szNeeded;
    register char *p;
    register int rLen;

    if (s->type == NST_FREE)
        return true;    // Possible due to multiple non coalesced events by evdp.
    //  simply ignore this call returning positive result.

    // Initialize p and szNeeded depending on change
    //
    switch (s->read.state) {
        case NRS_WAITOP:
            szNeeded = s->read.head_left;
            p = ((char *)&s->read.head[0]) + (2-szNeeded);
            break;

        case NRS_WAITLEN:
            szNeeded = s->read.head_left;
            p = ((char *)&s->read.head[1]) + (2-szNeeded);
            break;

        case NRS_WAITDATA: {
                register netbuf buf = s->read.buf;

                szNeeded = (buf->dataLen - buf->dataPos);
                p = (char *)&buf->buf[ buf->dataPos ];
            }
            break;

        default:
            // .. the impossible gets possible ..
            ShowError("_onRORecv: fd #%u has unknown read.state (%d) - disconnecting\n", fd, s->read.state);
            return false;
            break;
    }


    //

    rLen = read(fd, p, szNeeded);
    if (rLen == 0) {
        // eof..
        return false;
    } else if (rLen == -1) {

        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // try again later .. (this case shouldnt happen, because we're event trigered.. but .. sometimes it happens :)
            return true;
        }

        // an additional interesting case would be
        // EINTR, this 'could' be handled .. but:
        //  posix says that its possible that data gets currupted during irq
        //  or data gor read and not reported.., so we'd have a data loss..
        //  (which shouldnt happen with stream based protocols such as tcp)
        // its better to disonnect the client in that case.

        return false;
    }

    //
    // Got Data:
    //  next action also depends on current state ..
    //
    szNeeded -= rLen;
    switch (s->read.state) {
        case NRS_WAITOP:

            if (szNeeded > 0) {
                // still data missing ..
                s->read.head_left = szNeeded;
                return true; // wait for completion.
            } else {
                // complete ..
                //  next state depends on packet type.

                s->read.head[1] = ((uint16 *)s->netparser_data)[ s->read.head[0] ]; // store lenght of packet by opcode head[0] to head[1]

                if (s->read.head[1] == ROPACKET_UNKNOWN) {
                    // unknown packet - disconnect
                    ShowWarning("_onRORecv: fd #%u got unlnown packet 0x%04x - disconnecting.\n", fd, s->read.head[0]);
                    return false;
                } else if (s->read.head[1] == ROPACKET_DYNLEN) {
                    // dynamic length
                    // next state: requrie len.
                    s->read.state = NRS_WAITLEN;
                    s->read.head_left = 2;
                    return true; //
                } else if (s->read.head[1] == 2) {
                    // packet has no data (only opcode)
                    register netbuf buf = netbuffer_get(2); // :D whoohoo its giant!

                    NBUFW(buf, 0) = s->read.head[0]; // store opcode @ packet begin.
                    buf->dataPos = 2;
                    buf->dataLen = 2;
                    buf->next = NULL;

                    // Back to initial state -> Need opcode.
                    s->read.state = NRS_WAITOP;
                    s->read.head_left = 2;
                    s->read.buf = NULL;

                    // Call completion routine here.
                    s->onPacketComplete(fd,  s->read.head[0],  2,  buf);

                    return true; // done :)
                } else {
                    // paket needs .. data ..
                    register netbuf buf = netbuffer_get(s->read.head[1]);

                    NBUFW(buf, 0) = s->read.head[0]; // store opcode @ packet begin.
                    buf->dataPos = 2;
                    buf->dataLen = s->read.head[1];
                    buf->next = NULL;

                    // attach buffer.
                    s->read.buf = buf;

                    // set state:
                    s->read.state = NRS_WAITDATA;

                    return true;
                }

            }//endif: szNeeded > 0 (opcode read completed?)

            break;


        case NRS_WAITLEN:

            if (szNeeded > 0) {
                // incomplete ..
                s->read.head_left = szNeeded;
                return true;
            } else {

                if (s->read.head[1] == 4) {
                    // packet has no data (only opcode + length)
                    register netbuf buf = netbuffer_get(4);

                    NBUFL(buf, 0) = *((uint32 *)&s->read.head[0]); // copy  Opcode + length to netbuffer using MOVL
                    buf->dataPos = 4;
                    buf->dataLen = 4;
                    buf->next = NULL;

                    // set initial state (need opcode)
                    s->read.state = NRS_WAITOP;
                    s->read.head_left = 2;
                    s->read.buf = NULL;

                    // call completion routine.
                    s->onPacketComplete(fd,  s->read.head[0],  4,  buf);

                    return true;
                } else if (s->read.head[1] < 4) {
                    // invalid header.
                    ShowWarning("_onRORecv: fd #%u invalid header - got packet 0x%04x, reported length < 4 - INVALID - disconnecting\n", fd, s->read.head[0]);
                    return false;
                } else {
                    // Data needed
                    // next state -> waitdata!
                    register netbuf buf = netbuffer_get(s->read.head[1]);

                    NBUFL(buf, 0) = *((uint32 *)&s->read.head[0]); // copy  Opcode + length to netbuffer using MOVL
                    buf->dataPos = 4;
                    buf->dataLen = s->read.head[1];
                    buf->next = NULL;

                    // attach to session:
                    s->read.buf = buf;
                    s->read.state = NRS_WAITDATA;

                    return true;
                }

            }//endif: szNeeded > 0 (length read complete?)

            break;


        case NRS_WAITDATA:

            if (szNeeded == 0) {
                // Packet finished!
                // compltion.
                register netbuf buf = s->read.buf;

                // set initial state.
                s->read.state = NRS_WAITOP;
                s->read.head_left = 2;
                s->read.buf = NULL;

                // Call completion routine.
                s->onPacketComplete(fd,  NBUFW(buf, 0),  buf->dataLen,  buf);

                return true;
            } else {
                // still data needed
                s->read.buf->dataPos += rLen;

                return true;
            }
            break;


            //
        default:
            ShowError("_onRORecv: fd #%u has unknown read.state (%d) [2] - disconnecting\n", fd, s->read.state);
            return false;
            break;
    }


    return false;
}//end: _onRORecv()


void network_send(int32 fd,  netbuf buf)
{
    register SESSION *s = &g_Session[fd];

#ifdef PARANOID_CHECKS
    if (fd >= MAXCONN) {
        ShowError("network_send: tried to attach buffer to connection idientifer #%u which is out of bounds.\n", fd);
        _network_free_netbuf_async(buf);
        return;
    }
#endif


    if (s->type == NST_FREE)
        return;

    // Check Max Outstanding buffers limit.
    if ((s->write.max_outstanding > 0)   &&
        (s->write.n_outstanding >= s->write.max_outstanding)) {

        ShowWarning("network_send: fd #%u max Outstanding buffers exceeded. - disconnecting.\n", fd);
        network_disconnect(fd);
        //
        _network_free_netbuf_async(buf);
        return;
    }


    // Attach to the end:
    buf->next = NULL;
    if (s->write.buf_last != NULL) {
        s->write.buf_last->next = buf;
        s->write.buf_last = buf;

    } else {
        // currently no buffer attached.
        s->write.buf = s->write.buf_last = buf;

        // register @ evdp for writable notification.
        evdp_writable_add(fd, &s->evdp_data); //
    }


    //
    s->write.n_outstanding++;

}//end: network_send()


void network_parser_set_ro(int32 fd,
                           int16 *packetlentable,
                           void (*onPacketCompleteProc)(int32 fd,  uint16 op,  uint16 len,  netbuf buf)
                          )
{
    register SESSION *s = &g_Session[fd];
    register netbuf b, nb; // used for potential free attached buffers.

    if (s->type == NST_FREE)
        return;

    s->onPacketComplete = onPacketCompleteProc;

    s->onRecv = _onRORecv; // ..
    s->onSend = _onSend; // Using the normal generic netbuf based send function.

    s->netparser_data = packetlentable;

    // Initial State -> Need Packet OPCode.
    s->read.state = NRS_WAITOP;
    s->read.head_left = 2;


    // Detach (if..) all buffers.
    if (s->read.buf != NULL) {
        _network_free_netbuf_async(s->read.buf); //
        s->read.buf = NULL;
    }

    if (s->write.buf != NULL) {
        b = s->write.buf;
        while (1) {
            nb = b->next;

            _network_free_netbuf_async(b);

            b = nb;
        }

        s->write.buf = NULL;
        s->write.buf_last = NULL;
        s->write.n_outstanding = 0;
    }

    // not changing any limits on outstanding ..
    //

}//end: network_parser_set_ro()