BodgeMaster
/
NORM-mirror
mirror of https://github.com/USNavalResearchLaboratory/norm.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
NORM-mirror/examples/normSocket.cpp

1737 lines
71 KiB
C++
Raw Blame History

#include "normSocket.h"
#include <stdio.h> // for stderr
#include <assert.h> // for assert()
#include <string.h> // for strlen()
#include "protoTree.h"
#include "protoAddress.h"
#ifdef WIN32
#include <Winsock2.h> // for inet_ntoa() (TBD - change to use Protolib routines?)
#include <Ws2tcpip.h> // for inet_ntop()
#else
#include <arpa/inet.h> // for inet_ntoa() (TBD - change to use Protolib routines?)
#endif // if/else WIN32/UNIX
// COMPILE: (assumes "normApi.h" in "include" ...
// g++ -I../include -c normSocket.cpp
#define TRACE(...) fprintf(stderr, __VA_ARGS__)
// Extra, non-public NORM API functions used by NormSocket stuff
extern void NormSetId(NormSessionHandle sesssionHandle, NormNodeId normId);
// 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;
// This is extra stuff defined for NormSocket API extension purposes. As the NormSocket
// extension is finalized, these may be refined/relocated
enum {NORM_SOCKET_VERSION = 1};
enum NormSocketCommand
{
NORM_SOCKET_CMD_NULL = 0, // reserved, invalid/null command
NORM_SOCKET_CMD_REJECT, // sent by server-listener to reject invalid connection messages
NORM_SOCKET_CMD_ALIVE // TBD - for NormSocket "keep-alive" option?
};
// Default socket option values. Can be overrided with NormSetSocketOptions()
const UINT16 DEFAULT_NUM_DATA = 32;
const UINT16 DEFAULT_NUM_PARITY = 4;
const UINT16 DEFAULT_NUM_AUTO = 0;
const UINT16 DEFAULT_SEGMENT_SIZE = 1400;
const unsigned int DEFAULT_BUFFER_SIZE = 2*1024*1024;
// 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 NormSocketInfo : public ProtoTree::Item
{
public:
NormSocketInfo(unsigned int remoteAddrLen, const char* remoteAddr, UINT16 remotePort)
: norm_socket(NULL)
{
info_keysize = MakeKey(info_key, remoteAddrLen, remoteAddr, remotePort);
}
// copy constructor
NormSocketInfo(const NormSocketInfo& s)
{*this = s;}
~NormSocketInfo() {}
void SetSocket(class NormSocket* theSocket)
{norm_socket = theSocket;}
class NormSocket* GetSocket()
{return norm_socket;}
static unsigned int MakeKey(unsigned char* key, unsigned int remoteAddrLen, const char* remoteAddr, UINT16 remotePort)
{
key[0] = remoteAddrLen;
memcpy(key + 1, remoteAddr, remoteAddrLen);
unsigned int keysize = remoteAddrLen + 1;
memcpy(key + keysize, &remotePort, 2);
keysize += 2;
keysize <<= 3; // to size in 'bits'
return keysize;
}
void GetRemoteAddress(ProtoAddress& theAddr) const
{
int remoteAddrLen = info_key[0];
const char* remoteAddrPtr = (char*)info_key + 1;
ProtoAddress::Type addrType;
switch (remoteAddrLen)
{
case 4:
addrType = ProtoAddress::IPv4;
break;
case 16:
addrType = ProtoAddress::IPv6;
break;
default:
theAddr.Invalidate();
ASSERT(0);
return;
}
theAddr.SetRawHostAddress(addrType, remoteAddrPtr, remoteAddrLen);
UINT16 remotePort;
memcpy(&remotePort, remoteAddrPtr + remoteAddrLen, 2);
theAddr.SetPort(remotePort);
}
const char* GetKey() const
{return (const char*)info_key;}
unsigned int GetKeysize() const
{return info_keysize;}
private:
// remoteAddrLen + remoteAddr + remotePort
// 1 + 16 max + 2
unsigned char info_key[19];
unsigned int info_keysize;
class NormSocket* norm_socket; // may be NULL if it is pending acceptance
}; // end class NormSocketInfo
// helper function
NormSocketInfo NormGetSocketInfo(NormNodeHandle client)
{
char remoteAddr[16]; // big enough for IPv6
unsigned int remoteAddrLen = 16;
UINT16 remotePort;
NormNodeGetAddress(client, remoteAddr, &remoteAddrLen, &remotePort);
return NormSocketInfo(remoteAddrLen, remoteAddr, remotePort);
} // end NormGetSocketInfo()
class NormSocketTable : public ProtoTreeTemplate<NormSocketInfo>
{
public:
NormSocketInfo* FindSocketInfo(UINT16 remotePort, unsigned int remoteAddrLen, const char* remoteAddr)
{
unsigned char key[19];
unsigned int keysize = NormSocketInfo::MakeKey(key, remoteAddrLen, remoteAddr, remotePort);
return Find((char*)key, keysize);
}
NormSocketInfo* FindSocketInfo(NormNodeHandle client)
{
NormSocketInfo socketInfo = NormGetSocketInfo(client);
return Find(socketInfo.GetKey(), socketInfo.GetKeysize());
}
void RemoveSocketInfo(NormSocketInfo& socketInfo)
{
// safety dance
if (NULL == Find(socketInfo.GetKey(), socketInfo.GetKeysize())) return; // not on the dance floor
Remove(socketInfo);
}
}; // end class NormSocketTable
class NormSocket
{
public:
NormSocket(NormSessionHandle normSession = NORM_SESSION_INVALID);
~NormSocket();
// 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);}
NormSocket* GetServerSocket()
{return server_socket;}
NormInstanceHandle GetInstance() const
{return NormGetInstance(norm_session);}
NormSessionHandle GetSession() const
{return norm_session;}
NormSessionHandle GetMulticastSession() const
{return mcast_session;}
NormObjectHandle GetTxStream() const
{return tx_stream;}
void InitRxStream(NormObjectHandle rxStream)
{rx_stream = rxStream;}
NormObjectHandle GetRxStream() const
{return rx_stream;}
void SetFlowControl(bool state)
{
tx_flow_control = state;
if (NORM_OBJECT_INVALID != tx_stream)
NormStreamSetPushEnable(tx_stream, state ? false : true);
}
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;
tx_ready = true;
NormStreamSetPushEnable(tx_stream, tx_flow_control ? false : true);
}
bool Open(NormInstanceHandle instance = NORM_INSTANCE_INVALID);
bool Listen(UINT16 serverPort, const char* groupAddr, const char* serverAddr);
NormSocket* Accept(NormNodeHandle client, NormInstanceHandle instance = NORM_INSTANCE_INVALID);
bool Connect(const char* serverAddr, UINT16 serverPort, UINT16 localPort, 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 GetOptions(NormSocketOptions* options)
{
if (NULL != options) *options = socket_option;
}
bool SetOptions(NormSocketOptions* options)
{
// TBD - do validity checking and perhaps reset to defaults if (NULL == options)
if (NULL != options) socket_option = *options;
return true;
}
void SetSocketInfo(NormSocketInfo* socketInfo) // for server-side, client sockets only
{socket_info = socketInfo;}
NormSocketInfo* FindSocketInfo(NormNodeHandle client)
{return client_table.FindSocketInfo(client);}
void RemoveSocketInfo(NormSocketInfo& socketInfo) // for server sockets only
{client_table.RemoveSocketInfo(socketInfo);}
void SetUserData(const void* userData)
{user_data = userData;}
const void* GetUserData() const
{return user_data;}
void SetTrace(bool state);
void GetSocketEvent(const NormEvent& event, NormSocketEvent& socketEvent);
typedef enum State
{
CLOSED,
OPEN,
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);}
UINT16 GetLocalPort() const
{return (NORM_SESSION_INVALID != norm_session) ? NormGetRxPort(norm_session) : 0;}
//bool GetLocalAddress(char* addr, unsigned int& addrLen, UINT16& port)
// {return NormGetRxBindAddress(norm_session, addr, addrLen, port)}
void GetPeerName(char* addr, unsigned int* addrLen, UINT16* port)
{
if (NULL == addrLen) return;
switch (remote_version)
{
case 4:
if ((*addrLen >= 4) && (NULL != addr))
memcpy(addr, remote_addr, 4);
*addrLen = 4;
break;
case 6:
if ((*addrLen >= 16) && (NULL != addr))
memcpy(addr, remote_addr, 16);
*addrLen = 16;
break;
default:
*addrLen = 0;
return;
}
if (NULL != port) *port = remote_port;
}
private:
void UpdateRemoteAddress()
{
unsigned int addrLen = 16;
NormNodeGetAddress(remote_node, remote_addr, &addrLen, &remote_port);
if (4 == addrLen)
remote_version = 4;
else
remote_version = 6;
}
NormSocketOptions socket_option;
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
NormSocketTable client_table; // only applies to server sockets
NormSocketInfo* socket_info; // only applies to server-side, client sockets
unsigned int client_count; // only applies to mcast server sockets
NormNodeId client_id; // only applies to mcast client socket
NormNodeHandle remote_node; // client socket peer info
UINT8 remote_version; // 4 or 6
char remote_addr[16]; // big enough for IPv6
UINT16 remote_port;
// 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;
bool tx_flow_control;
// Receive stream state
NormObjectHandle rx_stream;
const void* user_data; // for use by user application
}; // end class NormSocket
NormSocket::NormSocket(NormSessionHandle normSession)
: socket_state(CLOSED), norm_session(normSession),
mcast_session(NORM_SESSION_INVALID), server_socket(NULL),
socket_info(NULL), client_count(0), client_id(NORM_NODE_NONE),
remote_node(NORM_NODE_INVALID), remote_version(0), remote_port(0),
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),
tx_flow_control(true),
rx_stream(NORM_OBJECT_INVALID), user_data(NULL)
{
// Initialize socket_option with default values
socket_option.num_data = DEFAULT_NUM_DATA;
socket_option.num_parity = DEFAULT_NUM_PARITY;
socket_option.num_auto = DEFAULT_NUM_AUTO;
socket_option.segment_size = DEFAULT_SEGMENT_SIZE;
socket_option.buffer_size = DEFAULT_BUFFER_SIZE;
socket_option.silent_receiver = false;
socket_option.max_delay = -1;
// 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);
}
NormSocket::~NormSocket()
{
Close();
if (NORM_SESSION_INVALID != norm_session)
{
NormDestroySession(norm_session);
norm_session = NORM_SESSION_INVALID;
}
}
bool NormSocket::Open(NormInstanceHandle instance)
{
if (CLOSED != socket_state)
{
fprintf(stderr, "NormSocket::Open() error: socket already open?!\n");
return false;
}
// A proper NormNodeId will be set upon NormBind(), NormConnect(), or NormListen()
if (NORM_SESSION_INVALID == (norm_session = NormCreateSession(instance, "127.0.0.1", 0, NORM_NODE_ANY)))
{
perror("NormSocket::Open() error");
return false;
}
NormSetUserData(norm_session, this);
socket_state = OPEN;
return true;
} // end NormSocket::Open()
bool NormSocket::Listen(UINT16 serverPort, const char* groupAddr, const char* serverAddr)
{
if (OPEN != socket_state)
{
/* This wasn't a good idea (yet and maybe never)
if ((CLOSED == socket_state) && (NORM_SESSION_INVALID != norm_session))
{
// closed socekt, not in use, so re-open socket ..
NormInstanceHandle instance = NormGetInstance(norm_session);
NormSessionHandle oldSession = norm_session;
if (!Open(instance))
{
norm_session = oldSession;
perror("NormSocket::Listen() error: unable to reopen socket");
return false;
}
else
{
NormDestroySession(oldSession);
}
}
else*/
{
fprintf(stderr, "NormSocket::Listen() error: socket not open!?\n");
return false;
}
}
// The code below will be cleaned/tightened up somewhat once all is working
// Note that 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"
if (NULL != groupAddr)
{
// TBD - validate that "groupAddr" is indeed a multicast address
NormChangeDestination(norm_session, groupAddr, serverPort);
NormSetId(norm_session, 1); // server always uses NormNodeId '1'
// TBD - we _could_ let the server have an independent, ephemeral tx_port
// by _not_ calling NormSetTxPort() here to enable multiple multicast
// servers on same group/port on same host if server instance use unique NormNodeIds?
NormSetTxPort(norm_session, serverPort); // can't do this and distinguish unicast feedback
NormSetMulticastInterface(norm_session, serverAddr);
NormSetRxPortReuse(norm_session, true);
mcast_session = norm_session;
NormSetMulticastLoopback(norm_session, true); // for testing
}
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
NormChangeDestination(norm_session, "127.0.0.1", serverPort);
NormSetId(norm_session, 1); // server always uses NormNodeId '1'
NormSetTxPort(norm_session, serverPort);
#ifdef WIN32
// UDP socket bind/connect does not work properly on WIN32, so no port reuse
// (so a little different strategy is used for Win32 connections)
NormSetRxPortReuse(norm_session, false, serverAddr);
#else
NormSetRxPortReuse(norm_session, true, serverAddr);
#endif // if/else WIN32
}
// TBD - the next four calls could be combined into a "NormStartListener()" function
// Set session to track incoming clients by their addr/port
// (instead of NormNodeId as usual)
NormSetServerListener(norm_session, true);
// Our listener is a "silent" receiver since all actual reception
// (unicast) is handed off to a separate "client" session
NormSetSilentReceiver(norm_session, true);
// So that the listener can construct (unsent) ACKs without failure
NormSetDefaultUnicastNack(norm_session, true);
// 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;
return false;
}
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);
#ifdef WIN32
NormSessionHandle clientSession = NormCreateSession(instance, clientAddr, 0, 1);
#else
NormSessionHandle clientSession = NormCreateSession(instance, clientAddr, serverPort, 1);
NormSetTxPort(clientSession, serverPort, true);
#endif // if/else WIN32/UNIX
// NORM_SYNC_STREAM tries to get everything the sender has cached/buffered
NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM);
//NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL);
// 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
#ifndef WIN32
// Enable rx port reuse since it's the server port, and connect
// this socket to client addr/port for unique, tight binding
// TBD - support option to bind to specific server address
//fprintf(stderr, "accepting connection from %s/%d on port %d ...\n", clientAddr, clientPort, serverPort);
NormSetRxPortReuse(clientSession, true, NULL, clientAddr, clientPort);
#endif // WIN32
NormSetDefaultUnicastNack(clientSession, true);
NormStartReceiver(clientSession, 2*1024*1024);
// This call immediately inserts the "client" remote sender state
// into the newly-created clientSession by injecting a NORM_CMD(CC)
// message (and, as a result, a NORM_ACK is sent back to the client
// for a quick initial RTT estimate)
NormTransferSender(clientSession, client);
NormSocket* clientSocket = new NormSocket(clientSession);
clientSocket->server_socket = this; // this is a server-side socket
clientSocket->remote_node = client;
clientSocket->UpdateRemoteAddress();
NormNodeSetUserData(client, clientSocket);
clientSocket->socket_option = socket_option; // inherit server_socket options (TBD - allow alt options passed into NormAccept()?)
NormNodeId clientId = NormNodeGetId(client);
if (IsUnicastSocket())
{
NormChangeDestination(clientSession, clientAddr, clientPort, false); // point unicast session dest to client port
// 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(),
socket_option.buffer_size, socket_option.segment_size,
socket_option.num_data, socket_option.num_parity);
NormSetAutoParity(clientSession, socket_option.num_auto);
}
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 = clientId;
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(),
socket_option.buffer_size, socket_option.segment_size,
socket_option.num_data, socket_option.num_parity);
NormSetAutoParity(norm_session, socket_option.num_auto);
socket_state = CONNECTED;
if (NORM_OBJECT_INVALID == tx_stream)
{
tx_stream = NormStreamOpen(norm_session,socket_option.buffer_size);
InitTxStream(tx_stream, socket_option.buffer_size, socket_option.segment_size, socket_option.num_data);
}
}
/* 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
*/
}
// Note that for this lightweight connection mode, ACCEPTING is essentially CONNECTED so
// app can treat this as a CONNECTED socket until otherwise notified
// Should we start the time clientSocket timer and timeout if not connected in a timely fashion?
clientSocket->socket_state = ACCEPTING; // will transition 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(const char* serverAddr,
UINT16 serverPort,
UINT16 localPort,
const char* groupAddr,
NormNodeId clientId)
{
if (OPEN != socket_state)
{
/* Not a good idea (yet and maybe never)
if ((CLOSED == socket_state) && (NORM_SESSION_INVALID != norm_session))
{
// closed socekt, not in use, so re-open socket ..
NormInstanceHandle instance = NormGetInstance(norm_session);
NormSessionHandle oldSession = norm_session;
if (!Open(instance))
{
norm_session = oldSession;
perror("NormSocket::Connect() error: unable to reopen socket");
return false;
}
else
{
NormDestroySession(oldSession);
}
}
else
*/
{
fprintf(stderr, "NormSocket::Connect() error: socket not open!?\n");
return false;
}
}
// 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 separate NormSession for multicast reception)
// Setting the session port to zero here causes an ephemeral port to be assigned _and_
// it is also a single socket (tx_socket == rx_socket) session for client->server unicast
NormSetId(norm_session, clientId);
// NORM_SYNC_STREAM tries to get everything the sender has cached/buffered
NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM);
//NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL);
#ifndef WIN32
// We don't set reuse for the ephemeral port, but do want to 'connect' to
// the server addr/port for this unicast client->server socket
NormSetRxPortReuse(norm_session, false, NULL, serverAddr, serverPort);
#endif // WIN32
if (0 != localPort)
{
// Set client session up to use a user-specified (non-ephemeral) port number
NormChangeDestination(norm_session, NULL, localPort, false);
NormSetTxPort(norm_session, localPort);
}
// 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;
}
NormSetSynStatus(norm_session, true);
// Point our unicast socket at the unicast server addr/port
NormChangeDestination(norm_session, serverAddr, serverPort, false);
NormSessionId sessionId = NormGetRandomSessionId(); // TBD - use ephemeral port as session/instance id?
//NormAddAckingNode(norm_session, 1); // servers always have NormNodeId '1' for unicast sessions
NormSetAutoAckingNodes(norm_session, NORM_TRACK_RECEIVERS); // this way we get informed upon first ACK
NormSetFlowControl(norm_session, 0); // since we do explicit, ACK-based flow control
if (!NormStartSender(norm_session, sessionId, 2*1024*1024, 1400, socket_option.num_data, socket_option.num_parity))
{
fprintf(stderr, "NormSocket::Connect() error: NormStartSender() failure\n");
return false;
}
NormSetAutoParity(norm_session, socket_option.num_auto);
if (NULL != groupAddr)
{
// Create the "mcast_session" for multicast reception
mcast_session = NormCreateSession(NormGetInstance(norm_session), 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(mcast_session, true); // for testing
client_id = clientId;
// TBD - make this SSM??? ... this would allow for multiple servers using the same groupAddr/port
// Note we 'connect' to server addr/port to make this associated with single, specific mcast server
// TBD - Once we add code to set multicast interface, we can set the port reuse
// here to 'connect' to the specified server addr/port for tighter binding
NormSetRxPortReuse(mcast_session, true, groupAddr);//, serverAddr, serverPort);
// 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
if (NORM_OBJECT_INVALID == tx_stream)
{
tx_stream = NormStreamOpen(norm_session, socket_option.buffer_size);
InitTxStream(tx_stream, socket_option.buffer_size, socket_option.segment_size, socket_option.num_data);
}
socket_state = CONNECTING;
return true;
} // end NormSocket::Connect()
unsigned int NormSocket::Write(const char* buffer, unsigned int numBytes)
{
// Make sure the socket is CONNECTED first
// (TBD - an option for allowing NormWrite() to start sending
// data prior to connection confirmation is being considered
// to accelerate data transfer (most useful for short-lived
// or 'urgent' connections such as transactions)
if (CONNECTED != socket_state) return 0;
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, socket_option.buffer_size);
InitTxStream(tx_stream, socket_option.buffer_size, socket_option.segment_size, socket_option.num_data);
}
if (!tx_flow_control)
{
unsigned int bytesWritten = NormStreamWrite(tx_stream, buffer, numBytes);
return bytesWritten;
}
else if (tx_stream_buffer_count < tx_stream_buffer_max)
{
// This method uses NormStreamWrite(), but limits writes by explicit ACK-based flow control status
// 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;
}
// TBD - set "tx_ready" to false if tx_stream_buffer_count == tx_stream_buffer_max here ???
return bytesWritten;
}
else
{
tx_ready = false;
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 rx-only client socket, so we redirect
// the flush() to the associated tx-only 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 ...
if (NORM_OBJECT_INVALID != rx_stream)
{
return NormStreamRead(rx_stream, buffer, &numBytes);
}
else
{
numBytes = 0;
return true;
}
} // end NormSocket::Read()
void NormSocket::Shutdown()
{
// TBD - should we call NormStopReceiver(norm_session) here
// or have SHUT_RD, SHUT_WR, and SHUT_RDWR flags
// like the sockets "shutdown()" call???
// For now, we do a "graceful" SHUT_RDWR behavior
if (CONNECTED == socket_state)
{
NormStopReceiver(norm_session);
rx_stream = NULL;
if ((IsServerSide() && IsMulticastClient()) || (NORM_OBJECT_INVALID == tx_stream))
{
// Use a zero-timeout to immediately post NORM_SOCKET_CLOSE notification
NormSetUserTimer(norm_session, 0.0);
}
else if (NORM_OBJECT_INVALID != tx_stream)
{
// 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;
}
else
{
// Use a zero-timeout to immediately post NORM_SOCKET_CLOSE notification
NormSetUserTimer(norm_session, 0.0);
}
} // 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
// timers so that NORM_SOCKET_CLOSE 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
{
NormDestroySession(mcast_session);
}
mcast_session = NORM_SESSION_INVALID;
}
if (NORM_SESSION_INVALID != norm_session)
{
NormCancelUserTimer(norm_session);
NormStopSender(norm_session);
NormStopReceiver(norm_session);
}
if (NULL != socket_info)
{
if (NULL != server_socket)
server_socket->RemoveSocketInfo(*socket_info);
delete socket_info;
socket_info = NULL;
}
// Iterate through remaining socket info and disassociate from any clients remaining
NormSocketTable::Iterator iterator(client_table);
NormSocketInfo* socketInfo;
while (NULL != (socketInfo = iterator.GetNextItem()))
{
NormSocket* clientSocket = socketInfo->GetSocket();
if (NULL != clientSocket)
clientSocket->SetSocketInfo(NULL);
client_table.Remove(*socketInfo);
delete socketInfo;
}
server_socket = NULL;
remote_node = NORM_NODE_INVALID;
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;
//fprintf(stderr, "NormSocket::GetSocketEvent() norm event type:%d session:%p\n", event.type, event.session);
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
socketEvent.type = NORM_SOCKET_CLOSE;
}
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_CLOSE;
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()
// (TBD - provide a NormSocket "keep alive" option
NormResetWatermark(event.session);
// Or just declare the connection broken/closed
//socketEvent.type = NORM_SOCKET_CLOSE;
}
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_CLOSE per failed client
// (This will also remove the client from this server's acking list)
clientSocket->socket_state = CLOSING;
NormSetUserTimer(clientSocket->norm_session, 0.0);
}
}
// 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_ACKING_NODE_NEW: // This means we have received an ACK from the server
case NORM_REMOTE_SENDER_RESET:
case NORM_REMOTE_SENDER_NEW:
{
switch (socket_state)
{
case LISTENING:
{
NormSocketInfo* socketInfo = client_table.FindSocketInfo(event.sender);
if (NULL == socketInfo)
{
// Add info for client socket pending acceptance
socketInfo = new NormSocketInfo(NormGetSocketInfo(event.sender));
if (NULL != socketInfo)
{
client_table.Insert(*socketInfo);
socketEvent.type = NORM_SOCKET_ACCEPT;
}
else
{
perror("NormSocket::GetSocketEvent() error: unable to add pending client info to server socket:\n");
}
}
else // duplicative accept event for existing socket, so ignore
{
ProtoAddress remoteAddr;
socketInfo->GetRemoteAddress(remoteAddr);
fprintf(stderr, "NormSocket::GetSocketEvent() warning: duplicative %s from client %s/%hu...\n",
(NORM_REMOTE_SENDER_NEW == event.type) ? "new" : "reset",
remoteAddr.GetHostString(), remoteAddr.GetPort());
// TBD - should we go ahead and delete this event.sender???
}
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;
NormSetSynStatus(norm_session, false);
socket_state = CONNECTED;
// Since UDP connect/bind doesn't really work properly on
// Windows, the Windows NormSocket server farms out client connections
// to new ephemeral port numbers, so we need to update
// the destination port upon connection (Yuck!)
remote_node = event.sender;
UpdateRemoteAddress();
NormChangeDestination(norm_session, NULL, remote_port);
if ((NORM_OBJECT_INVALID == tx_stream) && !(IsMulticastClient() && IsServerSide()))
{
tx_stream = NormStreamOpen(norm_session, socket_option.buffer_size);
InitTxStream(tx_stream, socket_option.buffer_size, socket_option.segment_size, socket_option.num_data);
}
break;
case CONNECTED:
if (IsMulticastSocket())
{
if (IsServerSocket())
{
// New client showing up at our multicast party
NormSocketInfo* socketInfo = client_table.FindSocketInfo(event.sender);
if (NULL == socketInfo)
{
// Add info for client socket pending acceptance
socketInfo = new NormSocketInfo(NormGetSocketInfo(event.sender));
if (NULL != socketInfo)
{
client_table.Insert(*socketInfo);
socketEvent.type = NORM_SOCKET_ACCEPT;
}
else
{
perror("NormSocket::GetSocketEvent() error: unable to add pending client info to server socket:\n");
}
}
else // duplicative accept event for existing socket, so ignore
{
ProtoAddress remoteAddr;
socketInfo->GetRemoteAddress(remoteAddr);
fprintf(stderr, "NormSocket::GetSocketEvent() warning: duplicative %s from client %s/%hu...\n",
(NORM_REMOTE_SENDER_NEW == event.type) ? "new" : "reset",
remoteAddr.GetHostString(), remoteAddr.GetPort());
// TBD - should we go ahead and delete this event.sender???
}
}
else
{
// TBD - validate if this same server or not (e.g. by source addr/port)
// Different sender showing up in multicast group!?
if (event.sender != remote_node)
{
char senderAddr[16];
unsigned int addrLen = 16;
UINT16 senderPort;
NormNodeGetAddress(event.sender, senderAddr, &addrLen, &senderPort);
unsigned int senderVersion = (4 == addrLen) ? 4 : 6;
if ((senderVersion != remote_version) ||
(senderPort != remote_port) ||
(0 != memcmp(senderAddr, remote_addr, addrLen)))
{
//fprintf(stderr, "NormSocket warning: multicast sender %s reset?!\n", NormNodeGetAddressString(event.sender));
}
}
// TBD - should Close() the socket and issue a NORM_SOCKET_CLOSE 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_CLOSE;
}
}
else // unicast
{
// Eemote sender reset? How do we tell?
// TBD - validate if this same server or not (e.g. by source addr/port)
if (event.sender != remote_node)
{
char senderAddr[16];
unsigned int addrLen = 16;
UINT16 senderPort;
NormNodeGetAddress(event.sender, senderAddr, &addrLen, &senderPort);
unsigned int senderVersion = (4 == addrLen) ? 4 : 6;
if ((senderVersion != remote_version) ||
(senderPort != remote_port) ||
(0 != memcmp(senderAddr, remote_addr, addrLen)))
{
fprintf(stderr, "NormSocket warning: unicast sender %s reset?!\n", NormNodeGetAddressString(event.sender));
}
}
// Close();
//socketEvent.type = NORM_SOCKET_CLOSE;
}
break;
default: // CLOSING, CLOSE
// shouldn't happen
break;
}
break;
}
case NORM_SEND_ERROR:
{
TRACE("NormSocket got SEND ERROR\n");
switch (socket_state)
{
case CONNECTING:
case ACCEPTING:
case CONNECTED:
case CLOSING:
if (IsMulticastServer())
fprintf(stderr, "SEND_ERROR on a multicast server socket?!\n");
/*else
fprintf(stderr, "SEND_ERROR session:%p sender:%p remote_node:%p (%s)\n",
event.session, event.sender, remote_node,
NormNodeGetAddressString(remote_node));*/
Close();
socketEvent.type = NORM_SOCKET_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:
Close();
socketEvent.type = NORM_SOCKET_CLOSE;
break;
default:
// shouldn't happen
assert(0);
break;
}
break;
}
case NORM_RX_CMD_NEW:
{
char buffer[4096];
unsigned int buflen = 4096;
if (NormNodeGetCommand(event.sender, buffer, &buflen))
{
if ((buflen < 2) || (NORM_SOCKET_VERSION == buffer[0]))
{
if (NORM_SOCKET_CMD_REJECT == buffer[1])
{
Close();
socketEvent.type = NORM_SOCKET_CLOSE;
}
else
{
fprintf(stderr, "NormSocket warning: received unknown command\n");
}
}
else
{
fprintf(stderr, "NormSocket warning: received command with invalid version\n");
}
}
else
{
fprintf(stderr, "NormSocket warning: unable to get received command\n");
}
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(NORM_RX_OBJECT_NEW) warning: 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:
// TBD - use an rx_ready indication to filter this event a little more
if (IsServerSocket()) break; // we don't receive data on server socket
if (event.object == rx_stream)
socketEvent.type = NORM_SOCKET_READ;
else
fprintf(stderr, "NormSocket::GetSocketEvent(NORM_RX_OBJECT_UPDATED) warning: non-matching rx object\n");
break;
default:
// shouldn't happen
break;
}
break;
}
case NORM_RX_OBJECT_ABORTED:
case NORM_RX_OBJECT_COMPLETED:
{
if (event.object != rx_stream)
break; // not our stream, so ignore
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
if (NORM_OBJECT_INVALID != tx_stream)
{
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;
}
else
{
// This still allows at least a chance of an ACK to be sent upon completion
NormSetUserTimer(norm_session, 0.0);
}
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;
}
break;
}
default:
break;
}
//fprintf(stderr, "NormSocket::GetSocketEvent() returning NormSocket event type:%d session:%p\n", socketEvent.type, event.session);
} // end NormSocket::GetSocketEvent()
void NormSocket::SetTrace(bool state)
{
if (NORM_SESSION_INVALID != norm_session)
NormSetMessageTrace(norm_session, state);
if (NORM_SESSION_INVALID != mcast_session)
NormSetMessageTrace(mcast_session, state);
} // end NormSocket::SetTrace()
///////////////////////////////////////////////////////////////////////////////////
// NormSocket API implementation
NormSocketHandle NormOpen(NormInstanceHandle instance)
{
NormSocket* normSocket = new NormSocket();
if (NULL == normSocket)
{
perror("NormOpen() new NormSocket() error");
return NORM_SOCKET_INVALID;
}
else if (normSocket->Open(instance))
{
return (NormSocketHandle)normSocket;
}
else
{
perror("NormOpen() error");
delete normSocket;
return NORM_SOCKET_INVALID;
}
} // end NormOpen()
// TBD - provide options for binding to a specific local address, interface, etc
bool NormListen(NormSocketHandle normSocket, UINT16 serverPort, const char* groupAddr, const char* serverAddr)
{
// TBD - make sure normSocket is valid
NormSocket* s = (NormSocket*)normSocket;
return s->Listen(serverPort, groupAddr, serverAddr);
} // end NormListen()
NormSocketHandle NormAccept(NormSocketHandle serverSocket, NormNodeHandle client, NormInstanceHandle instance)
{
// TBD - if another instance handle is provided use that instead
// TBD - VALIDATE PARAMETERS AND ERROR CHECK ALL THE API CALLS MADE HERE !!!!!
NormSocket* s = (NormSocket*)serverSocket;
NormInstanceHandle serverInstance = s->GetInstance();
NormSuspendInstance(serverInstance);
NormSocketHandle clientSocket = s->Accept(client, instance);
NormResumeInstance(serverInstance);
if (NORM_SOCKET_INVALID != clientSocket)
{
// Keep track of this client socket in our serverSocket socket_table
NormSocketInfo* socketInfo = s->FindSocketInfo(client);
ASSERT(NULL != socketInfo);
NormSocket* c = (NormSocket*)clientSocket;
socketInfo->SetSocket(c);
c->SetSocketInfo(socketInfo);
}
return clientSocket;
} // end NormAccept()
NORM_API_LINKAGE
extern bool NormSendCommandTo(NormSessionHandle sessionHandle,
const char* cmdBuffer,
unsigned int cmdLength,
const char* addr,
UINT16 port);
void NormReject(NormSocketHandle serverSocket,
NormNodeHandle clientNode)
{
// Simple, single "reject" command for moment (TBD - do something more stateful so app will be bothered less)
// Send "reject" command to source
char buffer[2];
buffer[0] = NORM_SOCKET_VERSION;
buffer[1] = NORM_SOCKET_CMD_REJECT;
NormSocket* s = (NormSocket*)serverSocket;
NormSocketInfo socketInfo = NormGetSocketInfo(clientNode);
ProtoAddress dest;
socketInfo.GetRemoteAddress(dest);
char destString[64];
destString[63] = '\0';
dest.GetHostString(destString, 63);
NormSendCommandTo(s->GetSession(), buffer, 2,destString, dest.GetPort());
} // end NormReject()
// TBD - provide options for binding to a specific local address, interface, etc
bool NormConnect(NormSocketHandle normSocket, const char* serverAddr, UINT16 serverPort, UINT16 localPort, const char* groupAddr, NormNodeId clientId)
{
// TBD - make sure normSocket is valid
NormSocket* s = (NormSocket*)normSocket;
NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance);
bool result = s->Connect(serverAddr, serverPort, localPort, groupAddr, clientId);
NormResumeInstance(instance);
return result;
} // 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;
NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance);
ssize_t result = (ssize_t)s->Write((const char*)buf, (unsigned int)nbyte);
NormResumeInstance(instance);
return result;
} // end NormWrite()
int NormFlush(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance);
s->Flush();
NormResumeInstance(instance);
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;
NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance);
// TBD - make sure s->rx_stream is valid
unsigned int numBytes = (unsigned int)nbyte;
ssize_t result;
if (s->Read((char*)buf, numBytes))
result = numBytes;
else
result = -1; // broken stream error (TBD - enumerate socket error values)
NormResumeInstance(instance);
return result;
} // end NormWrite()
void NormShutdown(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance);
s->Shutdown();
NormResumeInstance(instance);
} // end NormShutdown()
void NormClose(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance);
s->Close();
NormResumeInstance(instance);
delete s;
} // end NormClose()
void NormGetSocketOptions(NormSocketHandle normSocket, NormSocketOptions* options)
{
NormSocket* s = (NormSocket*)normSocket;
s->GetOptions(options);
} // end NormGetSocketOptions()
bool NormSetSocketOptions(NormSocketHandle normSocket, NormSocketOptions* options)
{
// TBD - do some validity checking, perhaps reset to defaults if (options == NULL)
NormSocket* s = (NormSocket*)normSocket;
return s->SetOptions(options);
} // end NormSetSocketOptions()
void NormSetSocketUserData(NormSocketHandle normSocket, const void* userData)
{
if (NORM_SOCKET_INVALID != normSocket)
((NormSocket*)normSocket)->SetUserData(userData);
} // end NormSetSocketUserData()
const void* NormGetSocketUserData(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
return s->GetUserData();
} // end NormGetSocketUserData()
// 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))
{
NormSuspendInstance(instance);
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);
}
NormResumeInstance(instance);
return true;
}
else
{
return false;
}
} // end NormGetSocketEvent()
// Other helper functions
void NormGetPeerName(NormSocketHandle normSocket, char* addr, unsigned int* addrLen, UINT16* port)
{
NormSocket* s = (NormSocket*)normSocket;
s->GetPeerName(addr, addrLen, port);
} // end NormGetPeerName()
NormSessionHandle NormGetSocketSession(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
return s->GetSession();
} // end NormGetSocketSession()
NormObjectHandle NormGetSocketTxStream(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
return s->GetTxStream();
} // end NormGetSocketTxStream()
NormObjectHandle NormGetSocketRxStream(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
return s->GetRxStream();
} // end NormGetSocketRxStream()
NormSessionHandle NormGetSocketMulticastSession(NormSocketHandle normSocket)
{
NormSocket* s = (NormSocket*)normSocket;
return s->GetMulticastSession();
} // end NormGetSocketMulticastSession()
void NormSetSocketTrace(NormSocketHandle normSocket, bool enable)
{
NormSocket* s = (NormSocket*)normSocket;
s->SetTrace(enable);
} // end NormSetSocketTrace()
void NormSetSocketFlowControl(NormSocketHandle normSocket, bool enable)
{
NormSocket* s = (NormSocket*)normSocket;
s->SetFlowControl(enable);
} // end NormSetSocketFlowControl()
Reference in New Issue View Git Blame Copy Permalink