norm-1.5r6/examples/normSocket.cpp

#include "normSocket.h"
#include <stdio.h>  // for stderr
#include <assert.h> // for assert()
#include <string.h>  // for strlen()
#include <arpa/inet.h>  // for inet_ntoa() (TBD - change to use Protolib routines?)

// COMPILE: (assumes "normApi.h" in "include" ...
// g++ -I../include -c normSocket.cpp


// This "NormSocket" class is used to maintain tx/rx state for a NORM "socket" connection.
// At the moment this "socket" connection represents a single, bi-directional NORM_OBJECT_STREAM
// in either a unicast context or an asymmetric "server" multicast stream to possibly multiple "client"
// nodes with individual unicast streams in return from those "client" nodes. (I.e., the server will need to
// have a normSocket per client even for the server multicast case (maybe :-) )

const NormSocketHandle NORM_SOCKET_INVALID = (NormSocketHandle)0;

const double NORM_DEFAULT_CONNECT_TIMEOUT = 60.0;

// a 'helper' function we use for debugging
const char* NormNodeGetAddressString(NormNodeHandle node)
{
    char addr[16];  // big enough for IPv6
    unsigned int addrLen = 16;
    UINT16 port;
    if (NormNodeGetAddress(node, addr, &addrLen, &port))
    {
        static char text[64];
        text[0] = text[31] = '\0';
        int addrFamily;
        if (4 == addrLen)
            addrFamily = AF_INET;
        else
            addrFamily = AF_INET6;
        inet_ntop(addrFamily, addr, text, 31);
        sprintf(text + strlen(text), "/%hu", port);
        return text;
    }
    else
    {
        return "???";
    }
}  // end NormNodeGetAddressString()

class NormSocket
{
    public:
        NormSocket(NormSessionHandle normSession = NORM_SESSION_INVALID);

        // These methods identify the role of this socket with respect
        // to the client / server relationship (a "server socket" is
        // one for which NormListen() has been invoked).
        bool IsServerSocket() const
            {return (server_socket == this);}
        bool IsClientSocket() const
            {return (server_socket != this);}
        bool IsUnicastSocket() const
            {return (NORM_SESSION_INVALID == mcast_session);}
        bool IsMulticastSocket() const
            {return !IsUnicastSocket();}
        bool IsMulticastServer() const
            {return (IsMulticastSocket() && IsServerSocket());}
        bool IsMulticastClient() const
            {return (IsMulticastSocket() && IsClientSocket());}
        bool IsServerSide() const
            {return (NULL != server_socket);}
        bool IsClientSide() const
            {return (NULL == server_socket);}

        NormSessionHandle GetSession() const
            {return norm_session;}
        NormSessionHandle GetMulticastSession() const
            {return mcast_session;}

        void InitRxStream(NormObjectHandle rxStream)
            {rx_stream = rxStream;}
        NormObjectHandle GetRxStream() const
            {return rx_stream;}

        void InitTxStream(NormObjectHandle txStream, unsigned int bufferSize, UINT16 segmentSize, UINT16 blockSize)
        {
            tx_stream = txStream;
            tx_segment_size = segmentSize;
            tx_stream_buffer_max = NormGetStreamBufferSegmentCount(bufferSize, segmentSize, blockSize);
            tx_stream_buffer_max -= blockSize;  // a little safety margin (perhaps not necessary)
            tx_stream_buffer_count = 0;
            tx_stream_bytes_remain = 0;
            tx_watermark_pending = false;
        }


        bool Listen(NormInstanceHandle instance, UINT16 serverPort, const char* groupAddr);
        NormSocket* Accept(NormNodeHandle client, NormInstanceHandle instance = NORM_INSTANCE_INVALID);
        bool Connect(NormInstanceHandle instance, const char* serverAddr, UINT16 serverPort, const char* groupAddr, NormNodeId clientId);


        // Write to tx stream (with flow control)
        unsigned int Write(const char* buffer, unsigned int numBytes);
        void Flush(bool eom = false, NormFlushMode flushMode = NORM_FLUSH_ACTIVE);
        // Read from rx_stream
        bool Read(char* buffer, unsigned int& numBytes);

        // "graceful" shutdown (stream is flushed and stream end, etc)
        void Shutdown();

        // hard, immediate closure
        void Close();

        void GetSocketEvent(const NormEvent& event, NormSocketEvent& socketEvent);

        typedef enum State
        {
            CLOSED,
            LISTENING,
            CONNECTING,
            ACCEPTING,
            CONNECTED,
            CLOSING
        } State;

        bool AddAckingNode(NormNodeId nodeId)
        {
            if (NormAddAckingNode(norm_session, nodeId))
            {
                client_count++;
                return true;
            }
            else
            {
                return false;
            }
        }
        void RemoveAckingNode(NormNodeId nodeId)
            {NormRemoveAckingNode(norm_session, nodeId);}


    private:
        State               socket_state;
        NormSessionHandle   norm_session;
        NormSessionHandle   mcast_session;   // equals norm_session for a multicast server
        NormSocket*         server_socket;   // only applies to server-side sockets
        unsigned int        client_count;    // only applies to mcast server sockets
        NormNodeId          client_id;       // only applies to mcast client socket
        NormNodeHandle      remote_node;     //
        // Send stream and associated flow control state variables
        NormObjectHandle    tx_stream;
        bool                tx_ready;
        UINT16              tx_segment_size;
        unsigned int        tx_stream_buffer_max;
        unsigned int        tx_stream_buffer_count;
        unsigned int        tx_stream_bytes_remain;
        bool                tx_watermark_pending;
        // Receive stream state
        NormObjectHandle    rx_stream;

};  // end class NormSocket


NormSocket::NormSocket(NormSessionHandle normSession)
 : socket_state(CLOSED), norm_session(normSession),
   mcast_session(NORM_SESSION_INVALID), server_socket(NULL),
   client_count(0), client_id(NORM_NODE_NONE), remote_node(NORM_NODE_INVALID),
   tx_stream(NORM_OBJECT_INVALID), tx_ready(false), tx_segment_size(0),
   tx_stream_buffer_max(0), tx_stream_buffer_count(0),
   tx_stream_bytes_remain(0), tx_watermark_pending(false),
   rx_stream(NORM_OBJECT_INVALID)
{
    // For now we use the NormSession "user data" option to associate
    // the session with a "socket".  In the future we may add a
    // dedicated NormSetSocket(NormSessionHandle session, NormSocketHandle normSocket) API
    // to keep the "user data" feature available for other purposes
    if (NORM_SESSION_INVALID != normSession) // this should always be true
        NormSetUserData(normSession, this);
}


bool NormSocket::Listen(NormInstanceHandle instance, UINT16 serverPort, const char* groupAddr)
{
    if (CLOSED != socket_state)
    {
        fprintf(stderr, "NormSocket::Listen() error: socket already open?!\n");
        return false;
    }

    if (NULL != groupAddr)
    {
        // TBD - validate that "groupAddr" is indeed a multicast address
        norm_session = NormCreateSession(instance, groupAddr, serverPort, NORM_NODE_ANY);
        NormSetTxPort(norm_session, serverPort); // can't do this and receive unicast feedback
        mcast_session = norm_session;
    }
    else
    {
        // For unicast , the "server" has a NormNodeId of '1' and the "clients" are '2'
        // to obviate need for explicit id management and will allow NAT to work, etc
        norm_session = NormCreateSession(instance, "127.0.0.1", serverPort, 1);
    }
    if (NORM_SESSION_INVALID == norm_session)
    {
        fprintf(stderr, "NormSocket::Listen() error: NormCreateSession() failure\n");
        return false;
    }
    NormSetUserData(norm_session, this);
    // Note the port reuse here lets us manage our "client" rx-only unicast connections the
    // way we need, but does allow a second multicast server to be started on this group which leads
    // to undefined behavior.  TBD - see if we can prevent via binding wizardry
    // (How is it done for TCP servers? - probably because the accept() call is in the network stack
    //  instead of user-space) Perhaps we could have a semaphore lock to block a second "server"
    NormSetRxPortReuse(norm_session, true);

    // use default sync policy so a "serversocket" doesn't NACK the senders it detects
    // NORM_SYNC_STREAM tries to get everything the sender has cached/buffered
    //NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM);
    //NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL);

    if (NULL == groupAddr)
    {
        // Unicast server
        // Note we use a small buffer size here since a "listening" socket isn't
        // going to be receiving data (TBD - implement a mechanism to handoff remote
        // sender (i.e. "client") from parent
        if (!NormStartReceiver(norm_session, 2048))
        {
            fprintf(stderr, "NormSocket::Listen() error: NormStartReceiver() failure (perhaps port already in use)\n");
            NormDestroySession(norm_session);
            norm_session = NORM_SESSION_INVALID;
        }
    }
    else
    {
        //NormSetMulticastInterface(norm_session, "lo0");
        NormSetMulticastLoopback(norm_session, true);  // for testing
        if (!NormStartReceiver(norm_session, 2048))
        {
            fprintf(stderr, "NormSocket::Listen() error: NormStartReceiver() failure (perhaps port already in use)\n");
            NormDestroySession(norm_session);
            norm_session = NORM_SESSION_INVALID;
        }
        // TBD - We _could_ go ahead and call NormStartSender(), but for now we'll wait until we hear the application
        //       makes at least one NormAccept() call ...
    }
    server_socket  = this;
    socket_state = LISTENING;
    return true;
}  // end NormSocket::Listen()

NormSocket* NormSocket::Accept(NormNodeHandle client, NormInstanceHandle instance)
{
    if (!IsServerSocket()) return NULL;
    char clientAddr[64];
    clientAddr[63] = '\0';
    char addr[16]; // big enough for IPv6
    unsigned int addrLen = 16;
    UINT16 clientPort;
    NormNodeGetAddress(client, addr, &addrLen, &clientPort);
    int addrFamily;
    UINT8 version;
    if (4 == addrLen)
    {
        addrFamily = AF_INET;
        version = 4;
    }
    else
    {
        addrFamily = AF_INET6;
        version = 6;
    }
    inet_ntop(addrFamily, addr, clientAddr, 63);

    UINT16 serverPort = NormGetRxPort(norm_session);
    if (NORM_INSTANCE_INVALID == instance)
        instance = NormGetInstance(norm_session);

    NormSessionHandle clientSession = NormCreateSession(instance, clientAddr, serverPort, 1);

    NormSetTxPort(clientSession, serverPort, false);

    // This next API call will cause NORM to tightly bind the remote client src addr/port to
    // our server port so the "clientSession" captures the client packets instead of the "server" session

    // Any new packets will come to our new connected clientSession instead
    // However, note that even though we've "connected" this sender,
    // there is a chance that additional packets in the "serverSession"
    // rx socket buffer may look like a new sender if deleted now, so
    // we wait for NORM_REMOTE_SENDER_INACTIVE to delete

    // NORM_SYNC_STREAM tries to get everything the sender has cached/buffered
    //NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM);
    NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL);

    NormSetRxPortReuse(clientSession, true, 0, clientAddr, clientPort);  // "connects" to remote client addr/port
    NormSetDefaultUnicastNack(clientSession, true);

    NormStartReceiver(clientSession, 2*1024*1024);

    NormSocket* clientSocket = new NormSocket(clientSession);
    clientSocket->server_socket = this;  // this is a server-side socket
    clientSocket->remote_node = client;
    NormNodeSetUserData(client, clientSocket);

    NormNodeId clientId = NormNodeGetId(client);

    if (IsUnicastSocket())
    {
        // The clientSession is bi-directional so we need to NormStartSender(), etc
        NormAddAckingNode(clientSession, 2); //clientId);
        NormSetFlowControl(clientSession, 0);  // disable timer-based flow control since we do explicit, ACK-based flow control
        NormStartSender(clientSession, NormGetRandomSessionId(), 2*1024*1024, 1400, 16, 4);
    }
    else // if IsMulticastSocket()
    {
        // TBD - should we make sure this not a NormNodeId we already have?
        // TBD - should we wait to add the client as acking node until CONNECT
        //       (probably for heavyweight; for lightweight we know the client
        //        has already started his multicast receiver)
        AddAckingNode(clientId);  // TBD - check result
        NormNodeHandle node = NormGetAckingNodeHandle(mcast_session, clientId);
        NormNodeSetUserData(node, clientSocket); // a way to track mcast client sockets
        clientSocket->mcast_session = mcast_session;
        clientSocket->client_id = client_id;
        if (LISTENING == socket_state)
        {
            NormSetFlowControl(norm_session, 0);  // disable timer-based flow control since we do explicit, ACK-based flow control
            NormStartSender(norm_session, NormGetRandomSessionId(), 2*1024*1024, 1400, 16, 4);
            socket_state = CONNECTED;
        }
        /* The code below would be invoked for "heavyweight" mcast client admission
          (for the moment we go with a "lightweight" model - this might be invokable upon
           as an optional behavior later)

        // Here, we start the clientSession (w/ a minimal buffer size) and create a temporary sender
        // stream that is immediately flushed/closed to inform the "client" that his connection
        // has been accepted.  The sender function is terminated upon client acknowledgement
        NormAddAckingNode(clientSession, clientId);
        NormSetFlowControl(clientSession, 0);  // disable timer-based flow control since we do explicit, ACK-based flow control
        NormStartSender(clientSession, NormGetRandomSessionId(), 1024, 512, 1, 0);
        NormObjectHandle tempStream = NormStreamOpen(clientSession, 1024);
        NormStreamClose(tempStream, true);  // Note our "trick" here to do a graceful close, _then_ watermark to get ack
        NormSetWatermark(clientSession, tempStream, true);  // future NORM API will add "bool watermark" option to graceful close
        */
    }
    clientSocket->socket_state = ACCEPTING;  // will transision to CONNECTED when client is detected on new clientSession
    return clientSocket;
}  // end NormSocket::Accept()

// TBD - provide options for binding to a specific local address, interface, etc
bool NormSocket::Connect(NormInstanceHandle instance, const char* serverAddr, UINT16 serverPort, const char* groupAddr, NormNodeId clientId)
{
    // For unicast connections, the "client" manages a single NormSession for send and receive
    // (For multicast connections, there are two sessions: The same unicast session that will
    //  be set to txOnly upon CONNECT and a NormSession for multicast reception)
    norm_session = NormCreateSession(instance, "127.0.0.1", 0, clientId);  // TBD - use "clientId" here for mcast sockets?
    if (NORM_SESSION_INVALID == norm_session)
    {
        fprintf(stderr, "NormSocket::Connect() error: NormCreateSession() failure\n");
        return false;
    }
    // NORM_SYNC_STREAM tries to get everything the sender has cached/buffered
    //NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM);
    NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL);

    NormSetUserData(norm_session, this);
    NormSetRxPortReuse(norm_session, true, NULL, serverAddr, serverPort);
    // TBD - for a multicast connection, the unicast receiver could be started with minimal buffer
    // (not that it matters since the buffers aren't activated until a sender starts sending _data_)
    if (!NormStartReceiver(norm_session, 2*1024*1024))  // to get ephemeral port assigned
    {
        fprintf(stderr, "NormSocket::Connect() error: unicast NormStartReceiver() failure\n");
        return false;
    }
    NormChangeDestination(norm_session, serverAddr, serverPort, false); // "connect" our NORM tx_socket (so we can get ICMP)
    NormSessionId sessionId = NormGetRandomSessionId();
    NormAddAckingNode(norm_session, 1);  // servers always have NormNodeId '1'
    NormSetFlowControl(norm_session, 0); // since we do explicit, ACK-based flow control
    if (!NormStartSender(norm_session, sessionId, 2*1024*1024, 1400, 16, 4))
    {
        fprintf(stderr, "NormSocket::Connect() error: NormStartSender() failure\n");
        return false;
    }

    if (NULL != groupAddr)
    {
        // Create the "mcast_session" for multicast reception
        mcast_session = NormCreateSession(instance, groupAddr, serverPort, clientId);
        //NormSetTxPort(mcast_session, serverPort);  // TBD - not sure this is a good idea if multiple clients on a machine?
        NormSetUserData(mcast_session, this);
        // NORM_SYNC_STREAM tries to get everything the sender has cached/buffered
        //NormSetDefaultSyncPolicy(mcast_session, NORM_SYNC_STREAM);
        NormSetDefaultSyncPolicy(mcast_session, NORM_SYNC_ALL);

        NormSetDefaultUnicastNack(mcast_session, true);  // we could optionally allow multicast NACKing, too
        NormSetMulticastLoopback(norm_session, true);  // for testing
        client_id = clientId;
        // TBD - make this SSM??? ... this would allow for multiple servers using the same groupAddr/port
        NormSetRxPortReuse(mcast_session, true, groupAddr);  // Should we upgrade rx port reuse and 'connect' to server tx port upon CONNECT?
        // For a "lightweight" client->server connection establishment, we go ahead and
        // stop our unicast receiver and start multicast receiver, assuming the server
        // will admit us into the group.
        // (TBD - provide a "heavier weight" connection acceptance confirmation/denial signal from server
        // via unicast from server -> client here (i.e. keep the unicast receiver open)
        if (!NormStartReceiver(mcast_session, 2*1024*1024))  // to get ephemeral port assigned
        {
            fprintf(stderr, "NormSocket::Connect() error: multicast NormStartReceiver() failure\n");
            return false;
        }
    }
    else
    {
        // Set timeout for connect attempt (for "heavyweight" mcast connect, this would also be done)
        NormSetUserTimer(norm_session, NORM_DEFAULT_CONNECT_TIMEOUT);
    }
    server_socket = NULL;  // this is a client-side socket

    socket_state = CONNECTING;

    return true;
}  // end NormSocket::Connect()


unsigned int NormSocket::Write(const char* buffer, unsigned int numBytes)
{
    // TBD - make sure the socket is CONNECTED first
    if (IsMulticastClient() && IsServerSide())
    {
        // This is multicast server rxonly client socket, so we redirect
        // the write() to the associated txonly multicast socket
        return server_socket->Write(buffer, numBytes);
    }
    // TBD - if tx_stream not yet open, open it!!!
    if (NORM_OBJECT_INVALID == tx_stream)
    {
        tx_stream = NormStreamOpen(norm_session, 2*1024*1024);
        InitTxStream(tx_stream, 2*1024*1024, 1400, 16);
    }

    // This method uses NormStreamWrite(), but limits writes by explicit ACK-based flow control status
    if (tx_stream_buffer_count < tx_stream_buffer_max)
    {
        // 1) How many buffer bytes are available?
        unsigned int bytesAvailable = tx_segment_size * (tx_stream_buffer_max - tx_stream_buffer_count);
        bytesAvailable -= tx_stream_bytes_remain;  // unflushed segment portiomn
        if (numBytes <= bytesAvailable)
        {
            unsigned int totalBytes = numBytes + tx_stream_bytes_remain;
            unsigned int numSegments = totalBytes / tx_segment_size;
            tx_stream_bytes_remain = totalBytes % tx_segment_size;
            tx_stream_buffer_count += numSegments;
        }
        else
        {
            numBytes = bytesAvailable;
            tx_stream_buffer_count = tx_stream_buffer_max;
        }
        // 2) Write to the stream
        unsigned int bytesWritten = NormStreamWrite(tx_stream, buffer, numBytes);
        //assert(bytesWritten == numBytes);  // this could happen if timer-based flow control is left enabled
        // 3) Check if we need to issue a watermark ACK request?
        if (!tx_watermark_pending && (tx_stream_buffer_count >= (tx_stream_buffer_max / 2)))
        {
            //fprintf(stderr, "tx_engine_t::WriteToNormStream() initiating watermark ACK request (buffer count:%lu max:%lu usage:%u)...\n",
            //            tx_stream_buffer_count, tx_stream_buffer_max, NormStreamGetBufferUsage(tx_stream));
            NormSetWatermark(norm_session, tx_stream);
            tx_watermark_pending = true;
        }
        return bytesWritten;
    }
    else
    {
        return 0;
    }
}  // end NormSocket::Write()

void NormSocket::Flush(bool eom, NormFlushMode flushMode)
{
    // TBD - make sure the socket is CONNECTED first
    if (IsMulticastClient() && IsServerSide())
    {
        // This is multicast server rxOnly client socket, so we redirect
        // the flush() to the associated txonly multicast socket
        return server_socket->Flush(eom, flushMode);
    }

    // NormStreamFlush always will transmit pending runt segments, if applicable
    // (thus we need to manage our buffer counting accordingly if pending bytes remain)
    if (tx_watermark_pending)
    {
        NormStreamFlush(tx_stream, eom, flushMode);
    }
    else if (NORM_FLUSH_ACTIVE == flushMode)
    {
        // we flush passive, because watermark forces active ack request
        NormStreamFlush(tx_stream, eom, NORM_FLUSH_PASSIVE);
        NormSetWatermark(norm_session, tx_stream, true);
    }
    else
    {
        NormStreamFlush(tx_stream, eom, flushMode);
    }

    if (0 != tx_stream_bytes_remain)
    {
        // The flush forces the runt segment out, so we increment our buffer usage count
        tx_stream_buffer_count++;
        tx_stream_bytes_remain = 0;
        if (!tx_watermark_pending && (tx_stream_buffer_count >= (tx_stream_buffer_max >> 1)))
        {
            //fprintf(stderr, "tx_engine_t::stream_flush() initiating watermark ACK request (buffer count:%lu max:%lu usage:%u)...\n",
            //       tx_stream_buffer_count, tx_stream_buffer_max);
            NormSetWatermark(norm_session, tx_stream, true);
            tx_watermark_pending = true;
        }
    }
 }  // end NormSocket::Flush()

 bool NormSocket::Read(char* buffer, unsigned int& numBytes)
 {
     // TBD - make sure rx_stream is valid!
     // TBD - make sure this is not a tx only client socket ...
     return NormStreamRead(rx_stream, buffer, &numBytes);
 }  // end NormSocket::Read()

 void NormSocket::Shutdown()
 {
     if ((NORM_OBJECT_INVALID == tx_stream)  ||
         (IsServerSide() && IsMulticastClient()))
     {
         Close();  // close immediately since this socket doesn't control a tx_stream
     }
     else
     {
         // It controls a tx_stream, so shutdown the tx_stream gracefully
         NormStreamClose(tx_stream, true);  // Note our "trick" here to do a graceful close, _then_ watermark to get ack
         NormSetWatermark(norm_session, tx_stream, true);  // future NORM API will add "bool watermark" option to graceful close
         socket_state = CLOSING;
     }
 }  // end NormSocket::Shutdown()

 void NormSocket::Close()
 {
     if (IsMulticastSocket())
     {
         if (IsServerSide())
         {
             if (IsServerSocket())
             {
                 // IsMulticastSocket() guarantees the mcast_session is valid
                 // Dissociate remaining clients from this session and set their
                 // timer so that NORM_SOCKET_CLOSED events are dispatched for them
                 NormNodeId nodeId = NORM_NODE_NONE;
                 while (NormGetNextAckingNode(mcast_session, &nodeId))
                 {
                     NormNodeHandle node = NormGetAckingNodeHandle(mcast_session, nodeId);
                     assert(NORM_NODE_INVALID != node);
                     NormSocket* clientSocket = (NormSocket*)NormNodeGetUserData(node);
                     NormSetUserTimer(clientSocket->norm_session, 0.0);
                 }
                 // for mcast server mcast_session == norm_session so it's destroyed below
             }
             else
             {
                 // "IsServerSide()" guarantees the "server_socket" is non-NULL
                 // server-side multicast client socket closing, so we
                 // need to remove this "client" NormNodeId from the mcast
                 // session's acking node list
                 server_socket->RemoveAckingNode(client_id);
            }
        }
        else  // client-side multicast socket, so we need to destroy mcast_session, too
        {
            NormDestroySession(mcast_session);
        }
        mcast_session = NORM_SESSION_INVALID;
     }
     if (NORM_SESSION_INVALID != norm_session)
     {
         NormDestroySession(norm_session);
         norm_session = NORM_SESSION_INVALID;
     }
     server_socket = NULL;
     tx_stream = NORM_OBJECT_INVALID;
     tx_segment_size = 0;
     tx_stream_buffer_max = tx_stream_buffer_count = tx_stream_bytes_remain = 0;
     tx_watermark_pending = false;
     rx_stream = NORM_OBJECT_INVALID;
     socket_state = CLOSED;
 }  // end NormSocket::Close()


void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketEvent)
{
    socketEvent.socket = (NormSocketHandle)this;
    socketEvent.type = NORM_SOCKET_NONE;  // default socket event type if no socket-specific state change occurs
    socketEvent.event = event;
    switch (event.type)
    {
        case NORM_TX_QUEUE_EMPTY:
        case NORM_TX_QUEUE_VACANCY:
        {
            // The socket may be tx ready, so issue a NORM_SOCKET_WRITE event
            if (CONNECTED == socket_state)
            {
                if (!tx_ready)
                {
                    tx_ready = true;
                    socketEvent.type = NORM_SOCKET_WRITE;
                }
            }
            break;
        }
        case NORM_TX_WATERMARK_COMPLETED:
        {
            switch (socket_state)
            {
                case ACCEPTING:
                {
                    // This only comes into play for the "confirmed connection"
                    // model for multicast sockets (not yet implemented)
                    assert(0);
                    assert(IsServerSide() && IsMulticastClient());
                    if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session))
                    {
                        // Client has acknowledged our acceptance
                        socketEvent.type = NORM_SOCKET_CONNECT;
                        NormStopSender(norm_session);  // the mcast_session is our tx channel
                        break;
                    }
                    else
                    {
                        // Client didn't acknowledge, so we cull him from our server
                        Close();
                        socketEvent.type = NORM_SOCKET_CLOSED;
                    }
                    break;
                }
                case CLOSING:
                {
                    // Socket that was shutdown has either been acknowledged or timed out
                    // TBD - should we issue a different event if ACK_FAILURE???
                    Close();
                    socketEvent.type = NORM_SOCKET_CLOSED;
                    break;
                }
                default:
                {
                    // TBD - implement option for more persistence
                    bool success = false;
                    if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session))
                    {
                        success = true;
                    }
                    else
                    {
                        // At least one receiver didn't acknowledge
                        if (IsUnicastSocket() || IsMulticastClient())
                        {
                            // We could be infinitely persistent w/ NormResetWatermark()
                            //NormResetWatermark(event.session);
                            // For now, we'll just declare the connection broken/closed
                            Close();
                            socketEvent.type = NORM_SOCKET_CLOSED;
                        }
                        else
                        {
                            // Multicast server, so determine who failed to acknowledge
                            // and cull them from our acking node list ... and shutdown
                            // their associated unicast sockets ... ugh!!!
                            NormNodeId nodeId = NORM_NODE_NONE;
                            NormAckingStatus ackingStatus;
                            while (NormGetNextAckingNode(mcast_session, &nodeId, &ackingStatus))
                            {
                                if (NORM_ACK_SUCCESS == ackingStatus)
                                {
                                    success = true;  // there was at least one success
                                }
                                else
                                {
                                    NormNodeHandle node = NormGetAckingNodeHandle(mcast_session, nodeId);
                                    assert(NORM_NODE_INVALID != node);
                                    NormSocket* clientSocket = (NormSocket*)NormNodeGetUserData(node);
                                    assert(NULL != clientSocket);
                                    // We use the session timer to dispatch a NORM_SOCKET_CLOSED per failed client
                                    // (This will also remove the client from this server's acking list)
                                    NormSetUserTimer(clientSocket->norm_session, 0.0);
                                    clientSocket->socket_state = CLOSING;
                                }
                            }
                            // TBD - what do we if all clients failed ... issue a NORM_SOCKET_DISCONNECT event,
                            // probably stop sending data and resume when a new client appears ???
                        }
                    }
                    if (tx_watermark_pending && success)
                    {
                        // flow control acknowledgement
                        tx_watermark_pending = false;
                        tx_stream_buffer_count -= (tx_stream_buffer_max >> 1);
                        if (!tx_ready)
                        {
                            tx_ready = true;
                            socketEvent.type = NORM_SOCKET_WRITE;
                        }
                    }
                    break;
                }
            }
            break;
        }
        case NORM_REMOTE_SENDER_RESET:
        case NORM_REMOTE_SENDER_NEW:
        {
            switch (socket_state)
            {
                case LISTENING:
                    socketEvent.type = NORM_SOCKET_ACCEPT;
                    break;
                case ACCEPTING:
                    if (IsServerSide() && IsClientSocket() && (NORM_NODE_INVALID != remote_node))
                        NormNodeDelete(remote_node);
                case CONNECTING:
                    // TBD - We should validate that it's the right remote sender
                    //       (i.e., by source address and/or nodeId)
                    NormCancelUserTimer(norm_session);
                    socketEvent.type = NORM_SOCKET_CONNECT;
                    socket_state = CONNECTED;
                    remote_node = event.sender;
                    break;
                case CONNECTED:
                    if (IsMulticastSocket())
                    {
                        if (IsServerSocket())
                        {
                            // New client showing up at our multicast party
                            socketEvent.type = NORM_SOCKET_ACCEPT;
                        }
                        else
                        {
                            // Different sender showing up in multicast group!?
                            fprintf(stderr, "NormSocket warning: multicast sender %s reset?!\n", NormNodeGetAddressString(event.sender));
                            // TBD - should Close() the socket and issue a NORM_SOCKET_CLOSED event
                            //        and leave it up to the application to reconnect?  Or should we
                            //        provides some sort of NORM_SOCKET_DISCONNECT event
                            socketEvent.type = NORM_SOCKET_CLOSED;
                            Close();
                        }
                    }
                    else  // unicast
                    {
                        // Eemote sender reset? How do we tell?
                        fprintf(stderr, "NormSocket warning: unicast sender %s reset?!\n", NormNodeGetAddressString(event.sender));
                        socketEvent.type = NORM_SOCKET_CLOSED;
                        Close();
                    }
                    break;

                default:  // CLOSING, CLOSED
                    // shouldn't happen
                    break;
            }
            break;
        }
        case NORM_SEND_ERROR:
        {
            switch (socket_state)
            {
                case CONNECTING:
                case ACCEPTING:
                case CONNECTED:
                case CLOSING:
                    if (IsMulticastServer())
                        fprintf(stderr, "SEND_ERROR on a multicast server socket?!\n");
                    socketEvent.event.sender = remote_node;
                    socketEvent.type = NORM_SOCKET_CLOSED;
                    Close();
                    break;
                default:
                    // shouldn't happen
                    break;
            }
            break;
        }

        case NORM_USER_TIMEOUT:
        {
            switch (socket_state)
            {
                case CONNECTING:    // client connection attempt timed out
                case ACCEPTING:     // accepted client didn't follow through
                case CONNECTED:     // multicast client ack failure
                case CLOSING:
                    socketEvent.event.sender = remote_node;
                    socketEvent.type = NORM_SOCKET_CLOSED;
                    Close();
                    break;
                default:
                    // shouldn't happen
                    assert(0);
                    break;
            }
            break;
        }
        case NORM_REMOTE_SENDER_INACTIVE:
        {
            switch (socket_state)
            {
                case LISTENING:
                {
                    // delete state for remote sender that has been accepted (or not)
                    // TBD - do something a little more tidy here
                    NormSocket* clientSocket = (NormSocket*)NormNodeGetUserData(event.sender);
                    if ((NULL != clientSocket) && (clientSocket->remote_node == event.sender))
                        clientSocket->remote_node = NORM_NODE_INVALID;
                    NormNodeDelete(event.sender);
                    break;
                }
                case CONNECTED:
                {
                    if (IsServerSocket())
                    {
                        NormSocket* clientSocket = (NormSocket*)NormNodeGetUserData(event.sender);
                        if ((NULL != clientSocket) && (clientSocket->remote_node == event.sender))
                            clientSocket->remote_node = NORM_NODE_INVALID;
                        NormNodeDelete(event.sender);
                    }
                    // TBD - should we do something here (perhaps issue a NORM_SOCKET_IDLE event or something
                    // that could be used as a clue that our "connection" may have broken or timed out???
                    // (Meanwhile, applications will have to figure that our for themselves)
                    break;
                }
                default:  // CONNECTING, ACCEPTING, CLOSING, CLOSED
                {
                    // shouldn't happen
                    break;
                }
            }
            break;
        }
        case NORM_RX_OBJECT_NEW:
        {
            switch (socket_state)
            {
                case LISTENING:
                    // TBD - shouldn't happen, delete sender right away?
                    break;
                case CONNECTED:
                    // TBD - make sure the sender is who we expect it to be???
                    if (IsServerSocket()) break;
                    if (NORM_OBJECT_INVALID == rx_stream)
                    {
                        // We're expecting this, new stream ready for reading ...
                        InitRxStream(event.object);
                        socketEvent.type = NORM_SOCKET_READ;
                    }
                    else
                    {
                        // Stream reset
                        fprintf(stderr, "NormSocket::GetSocketEvent(): client stream reset?!\n");
                    }
                    break;
                default:  // CONNECTING, ACCEPTING, CLOSING, CLOSED
                    // shouldn't happen
                    break;
            }
            break;
        }
        case NORM_RX_OBJECT_UPDATED:
        {
            switch (socket_state)
            {
                case CONNECTED:
                case CLOSING:  // we allow reading during graceful closure
                    // TBD - use an rx_ready indication to filter this event a little more
                    if (IsServerSocket()) break;  // we don't receive data on server socket
                    assert(event.object == rx_stream);
                    socketEvent.type = NORM_SOCKET_READ;
                    break;
                default:
                    // shouldn't happen
                    break;
            }
            break;
        }
        case NORM_RX_OBJECT_COMPLETED:
        {
            rx_stream = NORM_OBJECT_INVALID;
            switch (socket_state)
            {
                case CONNECTED:
                    // Initiate graceful closure of our tx_stream to allow at least some time to
                    // acknowledge the remote before closing everything down
                    NormStreamClose(tx_stream, true);  // Note our "trick" here to do a graceful close, _then_ watermark to get ack
                    NormSetWatermark(norm_session, tx_stream, true);  // future NORM API will add "bool watermark" option to graceful close
                    socket_state = CLOSING;
                    socketEvent.type = NORM_SOCKET_CLOSING;
                    break;

                case CLOSING:
                    // We're already closing, so just let that complete.  This helps make sure we allow
                    // at least some time to acknowledge the remote before closing everything down
                    break;
                default:
                    // shouldn't happen
                    break;
            }
        }
        default:
            break;
    }
}  // end NormSocket::GetSocketEvent()


 ///////////////////////////////////////////////////////////////////////////////////
 // NormSocket API implementation

// TBD - provide options for binding to a specific local address, interface, etc
NormSocketHandle NormListen(NormInstanceHandle instance, UINT16 serverPort, const char* groupAddr)
{
    // TBD - check results
    NormSocket* normSocket = new NormSocket();
    normSocket->Listen(instance, serverPort, groupAddr);
    return (NormSocketHandle)normSocket;
}  // end NormListen()


NormSocketHandle NormAccept(NormSocketHandle serverSocket, NormNodeHandle client)
{
    // TBD - VALIDATE PARAMETERS AND ERROR CHECK ALL THE API CALLS MADE HERE !!!!!
    NormSocket* s = (NormSocket*)serverSocket;
    return (NormSocketHandle)(s->Accept(client));
}  // end NormAccept()


// TBD - provide options for binding to a specific local address, interface, etc
NormSocketHandle NormConnect(NormInstanceHandle instance, const char* serverAddr, UINT16 serverPort, const char* groupAddr, NormNodeId clientId)
{
    NormSocket* normSocket = new NormSocket();
    if (NULL == normSocket)
    {
        perror("NormConnect() new NormSocket() error");
        return NULL;
    }
    if (normSocket->Connect(instance, serverAddr, serverPort, groupAddr, clientId))
    {
        return normSocket;
    }
    else
    {
        delete normSocket;
        return NULL;
    }
}  // end NormConnect()


ssize_t NormWrite(NormSocketHandle normSocket, const void *buf, size_t nbyte)
{
    // TBD - we could make write() and read() optionally blocking or non-blocking
    //       by using GetSocketEvent() as appropriate (incl. returning error conditions, etc)
    NormSocket* s = (NormSocket*)normSocket;
    return (ssize_t)s->Write((const char*)buf, nbyte);
}  // end NormWrite()

int NormFlush(NormSocketHandle normSocket)
{
    NormSocket* s = (NormSocket*)normSocket;
    s->Flush();
    return 0;
} // end NormFlush()

ssize_t NormRead(NormSocketHandle normSocket, void *buf, size_t nbyte)
{
    // TBD - we could make write() and read() optionally blocking or non-blocking
    //       by using GetSocketEvent() as appropriate (incl. returning error conditions, etc)
    NormSocket* s = (NormSocket*)normSocket;
    // TBD - make sure s->rx_stream is valid
    unsigned int numBytes = nbyte;
    if (s->Read((char*)buf, numBytes))
        return numBytes;
    else
        return -1; // broken stream error (TBD - enumerate socket error values)
}  // end NormWrite()


void NormShutdown(NormSocketHandle normSocket)
{
    NormSocket* s = (NormSocket*)normSocket;
    s->Shutdown();
}  // end NormShutdown()

void NormClose(NormSocketHandle normSocket)
{
    NormSocket* s = (NormSocket*)normSocket;
    s->Close();
}  // end NormClose()

// This gets and translates low level NORM API events to NormSocket events
// given the "normSocket" state
bool NormGetSocketEvent(NormInstanceHandle instance, NormSocketEvent* socketEvent, bool waitForEvent)
{
    if (NULL == socketEvent) return false;
    NormEvent event;
    if (NormGetNextEvent(instance, &event, waitForEvent))
    {
        NormSocket* normSocket = NULL;
        if (NORM_SESSION_INVALID != event.session)
            normSocket = (NormSocket*)NormGetUserData(event.session);
        if (NULL == normSocket)
        {
            socketEvent->type = NORM_SOCKET_NONE;
            socketEvent->socket = NORM_SOCKET_INVALID;
            socketEvent->event = event;
        }
        else
        {
            normSocket->GetSocketEvent(event, *socketEvent);
        }
        return true;
    }
    else
    {
        return false;
    }
}  // end NormGetSocketEvent()

// Other helper functions

NormSessionHandle NormGetSession(NormSocketHandle normSocket)
{
    NormSocket* s = (NormSocket*)normSocket;
    return s->GetSession();
}  // end NormGetSession()

NormSessionHandle NormGetMulticastSession(NormSocketHandle normSocket)
{
    NormSocket* s = (NormSocket*)normSocket;
    return s->GetMulticastSession();
}  // end NormGetSession()