detectors/slsDetectorPackage
25
0
Fork 0
mirror of https://github.com/slsdetectorgroup/slsDetectorPackage.git synced 2025-06-19 08:17:13 +02:00
Code Issues Actions Packages Releases Activity

Exceptions: zmq socket class descriptors made into its own class for auto destruction upon construction exception, similarly for other try blocks. slsDetector and multislsdetector left to do

Browse Source
This commit is contained in:
maliakal_d
2018-08-09 18:12:56 +02:00
parent 1102153d2b
commit a0512a01d5
7 changed files with 140 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
slsReceiverSoftware/include/ZmqSocket.h
View File

@ -45,12 +45,7 @@ public:
* @param portnumber port number
*/
ZmqSocket (const char* const hostname_or_ip, const uint32_t portnumber):
portno (portnumber),
server (false),
contextDescriptor (NULL),
socketDescriptor (NULL),
headerMessage(0)
{
portno (portnumber) {
char ip[MAX_STR_LENGTH] = "";
memset(ip, 0, MAX_STR_LENGTH);
@ -61,19 +56,19 @@ public:
throw std::exception();
// construct address
sprintf (serverAddress, "tcp://%s:%d", ip, portno);
sprintf (sockfd.serverAddress, "tcp://%s:%d", ip, portno);
#ifdef VERBOSE
cprintf(BLUE,"address:%s\n",serverAddress);
cprintf(BLUE,"address:%s\n",sockfd.serverAddress);
#endif
// create context
contextDescriptor = zmq_ctx_new();
if (contextDescriptor == NULL)
sockfd.contextDescriptor = zmq_ctx_new();
if (sockfd.contextDescriptor == 0)
throw std::exception();
// create publisher
socketDescriptor = zmq_socket (contextDescriptor, ZMQ_SUB);
if (socketDescriptor == NULL) {
sockfd.socketDescriptor = zmq_socket (sockfd.contextDescriptor, ZMQ_SUB);
if (sockfd.socketDescriptor == 0) {
PrintError ();
Close ();
throw std::exception();
@ -81,7 +76,7 @@ public:
//Socket Options provided above
// an empty string implies receiving any messages
if ( zmq_setsockopt(socketDescriptor, ZMQ_SUBSCRIBE, "", 0)) {
if ( zmq_setsockopt(sockfd.socketDescriptor, ZMQ_SUBSCRIBE, "", 0)) {
PrintError ();
Close();
throw std::exception();
@ -90,7 +85,7 @@ public:
//ZMQ_SNDHWM default is 0 means no limit. use this to optimize if optimizing required
// eg. int value = -1;
int value = 0;
if (zmq_setsockopt(socketDescriptor, ZMQ_LINGER, &value,sizeof(value))) {
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_LINGER, &value,sizeof(value))) {
PrintError ();
Close();
throw std::exception();
@ -105,19 +100,16 @@ public:
* @param ethip is the ip of the ethernet interface to stream zmq from
*/
ZmqSocket (const uint32_t portnumber, const char *ethip):
portno (portnumber),
server (true),
contextDescriptor (NULL),
socketDescriptor (NULL),
headerMessage(0)
{
portno (portnumber) {
sockfd.server = true;
// create context
contextDescriptor = zmq_ctx_new();
if (contextDescriptor == NULL)
sockfd.contextDescriptor = zmq_ctx_new();
if (sockfd.contextDescriptor == 0)
throw std::exception();
// create publisher
socketDescriptor = zmq_socket (contextDescriptor, ZMQ_PUB);
if (socketDescriptor == NULL) {
sockfd.socketDescriptor = zmq_socket (sockfd.contextDescriptor, ZMQ_PUB);
if (sockfd.socketDescriptor == 0) {
PrintError ();
Close ();
throw std::exception();
@ -126,12 +118,12 @@ public:
//Socket Options provided above
// construct addresss
sprintf (serverAddress,"tcp://%s:%d", ethip, portno);
sprintf (sockfd.serverAddress,"tcp://%s:%d", ethip, portno);
#ifdef VERBOSE
cprintf(BLUE,"address:%s\n",serverAddress);
cprintf(BLUE,"address:%s\n",sockfd.serverAddress);
#endif
// bind address
if (zmq_bind (socketDescriptor, serverAddress) < 0) {
if (zmq_bind (sockfd.socketDescriptor, sockfd.serverAddress) < 0) {
PrintError ();
Close ();
throw std::exception();
@ -145,69 +137,49 @@ public:
* Destructor
*/
~ZmqSocket () {
Disconnect();
Close();
//mySocketDescriptor destructor also gets called
};
/**
* Returns Server Address
* @returns Server Address
*/
char* GetZmqServerAddress () { return serverAddress; };
/**
* Returns Port Number
* @returns Port Number
*/
uint32_t GetPortNumber () { return portno; };
/**
* Returns Server Address
* @returns Server Address
*/
char* GetZmqServerAddress () { return sockfd.serverAddress; };
/**
* Returns Socket Descriptor
* @reutns Socket descriptor
*/
void* GetsocketDescriptor () { return socketDescriptor; };
void* GetsocketDescriptor () { return sockfd.socketDescriptor; };
/**
* Connect client socket to server socket
* @returns 1 for fail, 0 for success
*/
int Connect() {
if (zmq_connect(socketDescriptor, serverAddress) < 0) {
if (zmq_connect(sockfd.socketDescriptor, sockfd.serverAddress) < 0) {
PrintError ();
Close ();
return 1;
}
return 0;
}
/**
* Unbinds the Socket
*/
void Disconnect () {
if (server)
zmq_unbind (socketDescriptor, serverAddress);
else
zmq_disconnect (socketDescriptor, serverAddress);
};
void Disconnect () {sockfd.Disconnect();};
/**
* Close Socket and destroy Context
*/
void Close () {
if (socketDescriptor != NULL) {
zmq_close (socketDescriptor);
socketDescriptor = NULL;
}
if (contextDescriptor != NULL) {
zmq_ctx_destroy (contextDescriptor);
contextDescriptor = NULL;
}
};
void Close () { sockfd.Close(); };
/**
* Convert Hostname to Internet address info structure
@ -332,7 +304,7 @@ public:
cprintf(BLUE,"%d : Streamer: buf: %s\n", index, buf);
#endif
if(zmq_send (socketDescriptor, buf, length, dummy?0:ZMQ_SNDMORE) < 0) {
if(zmq_send (sockfd.socketDescriptor, buf, length, dummy?0:ZMQ_SNDMORE) < 0) {
PrintError ();
return 0;
}
@ -349,7 +321,7 @@ public:
* @returns 0 if error, else 1
*/
int SendData (char* buf, int length) {
if(zmq_send (socketDescriptor, buf, length, 0) < 0) {
if(zmq_send (sockfd.socketDescriptor, buf, length, 0) < 0) {
PrintError ();
return 0;
}
@ -367,7 +339,7 @@ public:
* @returns length of message, -1 if error
*/
int ReceiveMessage(const int index, zmq_msg_t& message) {
int length = zmq_msg_recv (&message, socketDescriptor, 0);
int length = zmq_msg_recv (&message, sockfd.socketDescriptor, 0);
if (length == -1) {
PrintError ();
cprintf (BG_RED,"Error: Could not read header for socket %d\n",index);
@ -461,29 +433,6 @@ public:
dummy = temp ? false : true;
return 1;
/*
int temp = d["data"].GetUint();
dummy = temp ? false : true;
if (!dummy) {
acqIndex = d["acqIndex"].GetUint64();
frameIndex = d["fIndex"].GetUint64();
fileIndex = d["fileIndex"].GetUint64();
subframeIndex = d["expLength"].GetUint();
filename = d["fname"].GetString();
}
#ifdef VERYVERBOSE
cprintf(BLUE,"%d Dummy:%d\n"
"\tAcqIndex:%lu\n"
"\tFrameIndex:%lu\n"
"\tSubIndex:%u\n"
"\tFileIndex:%lu\n"
"\tBitMode:%u\n"
"\tDetType:%u\n",
index, (int)dummy, acqIndex, frameIndex, subframeIndex, fileIndex,
d["bitmode"].GetUint(),d["detType"].GetUint());
#endif
return 1;
*/
};
@ -585,24 +534,60 @@ public:
};
private:
/**
* Class to close socket descriptors automatically
* upon encountering exceptions in the ZmqSocket constructor
*/
class mySocketDescriptors {
public:
/** Constructor */
mySocketDescriptors():
server(false),
contextDescriptor(0),
socketDescriptor(0) {};
/** Destructor */
~mySocketDescriptors() {
Disconnect();
Close();
}
/** Unbinds the Socket */
void Disconnect () {
if (server)
zmq_unbind (socketDescriptor, serverAddress);
else
zmq_disconnect (socketDescriptor, serverAddress);
};
/** Close Socket and destroy Context */
void Close () {
if (socketDescriptor != NULL) {
zmq_close (socketDescriptor);
socketDescriptor = NULL;
}
if (contextDescriptor != NULL) {
zmq_ctx_destroy (contextDescriptor);
contextDescriptor = NULL;
}
};
/** true if server, else false */
bool server;
/** Server Address */
char serverAddress[1000];
/** Context Descriptor */
void* contextDescriptor;
/** Socket Descriptor */
void* socketDescriptor;
};
private:
/** Port Number */
uint32_t portno;
/** true if server, else false */
bool server;
/** Context Descriptor */
void* contextDescriptor;
/** Socket Descriptor */
void* socketDescriptor;
/** Server Address */
char serverAddress[1000];
/** Socket descriptor */
mySocketDescriptors sockfd;
/** Header Message pointer */
zmq_msg_t* headerMessage;
};