able to send data to eigerbe

tweaks on logging

added configuration_map, to flexibly pass options to the different receivers

some more fixes to method overloads

still very preliminary
This commit is contained in:
sala 2014-09-17 15:56:27 +02:00
parent cfacc6ad6e
commit b427452f39
12 changed files with 352 additions and 117 deletions

View File

@ -19,17 +19,13 @@
#include "JsonBox/Value.h"
//#include "logger.h"
#include <iostream>
#include <sstream>
#include <string>
#include <exception>
#define EIGER_DEBUG
#ifdef EIGER_DEBUG
#define DEBUG(x) do { std::cerr << "[DEBUG] " << x << std::endl; } while (0)
#else
#define DEBUG(x)
#endif
using namespace Poco::Net;
@ -39,7 +35,7 @@ using namespace std;
class RestHelper {
public:
RestHelper(int timeout=10, int n_tries=3){
RestHelper(int timeout=10, int n_tries=10){
/**
*
*
@ -78,18 +74,22 @@ class RestHelper {
*/
//Check for http:// string
FILE_LOG(logDEBUG) << __func__ << " starting";
string proto_str = "http://";
if( size_t found = hostname.find(proto_str) != string::npos ){
cout << hostname << endl;
char c1[hostname.size()-found-1];
cout << c1 << endl;
size_t length1 = hostname.copy(c1, hostname.size()-found-1, proto_str.size());
c1[length1]='\0';
hostname = c1;
}
full_hostname = "http://"+hostname;
session = new HTTPClientSession(hostname,port );
session = new HTTPClientSession(hostname, port );
session->setKeepAliveTimeout( Timespan( http_timeout,0) );
};
@ -168,7 +168,7 @@ class RestHelper {
string answer;
int code = send_request(session, req, &answer);
if(code == 0 ) {
DEBUG("ANSWER " << answer );
FILE_LOG(logDEBUG) << "ANSWER " << answer;
json_value->loadFromString(answer);
}
delete uri;
@ -176,7 +176,7 @@ class RestHelper {
};
int post_json(string request, string *answer, string request_body=""){
int post_json(string request, string *answer, string request_body="{}"){
/**
*
*
@ -192,15 +192,16 @@ class RestHelper {
if (path.empty()) path = "/";
HTTPRequest req(HTTPRequest::HTTP_POST, path, HTTPMessage::HTTP_1_1 );
req.setContentType("application/json\r\n");
req.setContentLength( request.length() );
cout << "REQUEST BODY " << request_body << endl;
req.setContentLength( request_body.length() );
int code = send_request(session, req, answer, request_body);
delete uri;
return code;
}
int post_json(string request, JsonBox::Value* json_value, string request_body=""){
int post_json(string request, JsonBox::Value* json_value, string request_body="{}"){
/**
*
*
@ -283,7 +284,7 @@ class RestHelper {
return code;
}
catch (exception& e){
cout << "Exception connecting to "<< full_hostname << ": "<< e.what() << ", sleeping 5 seconds (" << n << "/"<<n_connection_tries << ")" << endl;
FILE_LOG(logERROR) << "Exception connecting to "<< full_hostname << ": "<< e.what() << ", sleeping 5 seconds (" << n << "/"<<n_connection_tries << ")";
sleep(5);
}
n+=1;

View File

@ -32,7 +32,7 @@
* @short does all the functions for a receiver, set/get parameters, start/stop etc.
*/
class UDPBaseImplementation : private virtual slsReceiverDefs, public UDPInterface {
class UDPBaseImplementation : protected virtual slsReceiverDefs, public UDPInterface {
public:
/**
@ -45,6 +45,7 @@ class UDPBaseImplementation : private virtual slsReceiverDefs, public UDPInterfa
*/
virtual ~UDPBaseImplementation();
void configure(map<string, string> config_map);
/**

View File

@ -13,6 +13,7 @@
* @short base class with all the functions for a receiver, set/get parameters, start/stop etc.
*/
#include <exception>
#include "sls_receiver_defs.h"
#include "receiver_defs.h"
@ -20,7 +21,6 @@
#include "utilities.h"
#include "logger.h"
/*
void print_not_implemented(string method_name){
std::cout << "[WARNING] Method " << method_name << " not implemented!" << std::endl;
@ -79,8 +79,9 @@ class UDPInterface {
*/
static UDPInterface *create(string receiver_type = "standard");
virtual void configure(map<string, string> config_map) = 0;
public:
public:

View File

@ -34,7 +34,7 @@
* @short does all the functions for a receiver, set/get parameters, start/stop etc.
*/
class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImplementation {
class UDPRESTImplementation : protected virtual slsReceiverDefs, public UDPBaseImplementation {
public:
/**
@ -48,6 +48,9 @@ class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImp
virtual ~UDPRESTImplementation();
void initialize_REST();
void configure(map<string, string> config_map);
/**
* delete and free member parameters
@ -57,14 +60,28 @@ class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImp
/**
* initialize member parameters
*/
void initializeMembers();
//void initializeMembers();
/**
* Set detector hostname
* @param c hostname
*/
//void initialize(const char *detectorHostName);
/* Returns detector hostname
/returns hostname
* caller needs to deallocate the returned char array.
* if uninitialized, it must return NULL
*/
//char *getDetectorHostname() const;
/**
* Set receiver type
* @param det detector type
* Returns success or FAIL
*/
int setDetectorType(detectorType det);
//int setDetectorType(detectorType det);
//Frame indices and numbers caught
@ -192,18 +209,6 @@ class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImp
*/
runStatus getStatus() const;
/**
* Set detector hostname
* @param c hostname
*/
void initialize(const char *detectorHostName);
/* Returns detector hostname
/returns hostname
* caller needs to deallocate the returned char array.
* if uninitialized, it must return NULL
*/
//char *getDetectorHostname() const;
/**
* Set Ethernet Interface or IP to listen to
@ -811,6 +816,9 @@ public:
//REST specific
bool isInitialized;
RestHelper * rest ;
int rest_port; // receiver backend port
string rest_hostname; // receiver hostname
};

View File

@ -5,6 +5,19 @@
#include <string>
#include <stdio.h>
#define STRINGIFY(x) #x
#define TOSTRING(x) STRINGIFY(x)
#define MYCONCAT(x,y)
#define __AT__ string(__FILE__) + string("::") + string(__func__) + string("(): ")
//":" TOSTRING(__LINE__)
/*
void error(const char *location, const char *msg){
printf("Error at %s: %s\n", location, msg);
}
*/
inline std::string NowTime();
enum TLogLevel {logERROR, logWARNING, logINFO, logDEBUG, logDEBUG1, logDEBUG2, logDEBUG3, logDEBUG4};
@ -110,7 +123,7 @@ template <typename T> std::ostringstream& Log<T>::Get(TLogLevel level)
template <typename T> Log<T>::~Log()
{
os << std::endl;
T::Output(os.str());
T::Output( os.str());
}
template <typename T> TLogLevel& Log<T>::ReportingLevel()

View File

@ -2,6 +2,8 @@
#include <string>
#include <sstream>
#include <iostream>
#include <map>
using namespace std;
#include "sls_receiver_defs.h"
@ -9,5 +11,5 @@ using namespace std;
//#define EIGER_DEBUG
int read_config_file(string fname, int *tcpip_port_no);
int read_config_file(string fname, int *tcpip_port_no, map<string, string> * configuration_map);

View File

@ -35,6 +35,11 @@ UDPBaseImplementation::UDPBaseImplementation(){}
UDPBaseImplementation::~UDPBaseImplementation(){}
void UDPBaseImplementation::configure(map<string, string> config_map){
FILE_LOG(logWARNING) << __AT__ << "doing nothing...";
};
void UDPBaseImplementation::deleteMembers(){
FILE_LOG(logWARNING) << "[WARNING] This is a base implementation, " << __func__ << " could have no effects.";
}
@ -47,6 +52,92 @@ void UDPBaseImplementation::initializeMembers(){
int UDPBaseImplementation::setDetectorType(detectorType det){
cout << "[WARNING] This is a base implementation, " << __func__ << " not correctly implemented" << endl;
cout << "Setting Receiver Type " << endl;
deleteMembers();
initializeMembers();
myDetectorType = det;
switch(myDetectorType){
case GOTTHARD:
cout << endl << "***** This is a GOTTHARD Receiver *****" << endl << endl;
break;
case MOENCH:
cout << endl << "***** This is a MOENCH Receiver *****" << endl << endl;
break;
case EIGER:
cout << endl << "***** This is a EIGER Receiver *****" << endl << endl;
break;
default:
cout << endl << "***** Unknown Receiver *****" << endl << endl;
return FAIL;
break;
}
//moench variables
if(myDetectorType == GOTTHARD){
fifosize = GOTTHARD_FIFO_SIZE;
packetsPerFrame = GOTTHARD_PACKETS_PER_FRAME;
onePacketSize = GOTTHARD_ONE_PACKET_SIZE;
frameSize = GOTTHARD_BUFFER_SIZE;
bufferSize = GOTTHARD_BUFFER_SIZE;
maxPacketsPerFile = MAX_FRAMES_PER_FILE * GOTTHARD_PACKETS_PER_FRAME;
frameIndexMask = GOTTHARD_FRAME_INDEX_MASK;
frameIndexOffset = GOTTHARD_FRAME_INDEX_OFFSET;
packetIndexMask = GOTTHARD_PACKET_INDEX_MASK;
}else if(myDetectorType == MOENCH){
fifosize = MOENCH_FIFO_SIZE;
packetsPerFrame = MOENCH_PACKETS_PER_FRAME;
onePacketSize = MOENCH_ONE_PACKET_SIZE;
frameSize = MOENCH_BUFFER_SIZE;
bufferSize = MOENCH_BUFFER_SIZE;
maxPacketsPerFile = MOENCH_MAX_FRAMES_PER_FILE * MOENCH_PACKETS_PER_FRAME;
frameIndexMask = MOENCH_FRAME_INDEX_MASK;
frameIndexOffset = MOENCH_FRAME_INDEX_OFFSET;
packetIndexMask = MOENCH_PACKET_INDEX_MASK;
}
else if(myDetectorType == EIGER){
fifosize = EIGER_FIFO_SIZE;
packetsPerFrame = EIGER_ONE_GIGA_CONSTANT * dynamicRange * EIGER_MAX_PORTS;
onePacketSize = EIGER_ONE_GIGA_ONE_PACKET_SIZE;
frameSize = onePacketSize * packetsPerFrame;
bufferSize = (frameSize/EIGER_MAX_PORTS) + EIGER_HEADER_LENGTH;//everything one port gets (img header plus packets)
maxPacketsPerFile = EIGER_MAX_FRAMES_PER_FILE * packetsPerFrame;
frameIndexMask = EIGER_FRAME_INDEX_MASK;
frameIndexOffset = EIGER_FRAME_INDEX_OFFSET;
packetIndexMask = EIGER_PACKET_INDEX_MASK;
pthread_mutex_lock(&status_mutex);
listeningthreads_mask = 0x0;
pthread_mutex_unlock(&(status_mutex));
if(thread_started)
createListeningThreads(true);
numListeningThreads = MAX_NUM_LISTENING_THREADS;
}
latestData = new char[frameSize];
setupFifoStructure();
if(createListeningThreads() == FAIL){
cout << "ERROR: Could not create listening thread" << endl;
exit (-1);
}
if(createWriterThreads() == FAIL){
cout << "ERROR: Could not create writer threads" << endl;
exit (-1);
}
setThreadPriorities();
cout << "Ready..." << endl;
return OK;
return OK;
}
@ -94,12 +185,11 @@ void UDPBaseImplementation::resetTotalFramesCaught(){
/*file parameters*/
char* UDPBaseImplementation::getFilePath() const{
//FILE_LOG(logWARNING) << "[WARNING] This is a base implementation, " << __func__ << " could have no effects.";
return (char*)filePath;
}
inline char* UDPBaseImplementation::setFilePath(const char c[]){
cout << "SET FILE PATH " << c << endl;
FILE_LOG(logDEBUG) << __AT__ << "called";
if(strlen(c)){
//check if filepath exists
struct stat st;
@ -110,6 +200,7 @@ inline char* UDPBaseImplementation::setFilePath(const char c[]){
FILE_LOG(logWARNING) << "FilePath does not exist:" << filePath;
}
}
FILE_LOG(logDEBUG) << __AT__ << getFilePath();
cout << getFilePath() << " " << filePath << endl;
return getFilePath();
}
@ -628,6 +719,7 @@ void UDPBaseImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char
int UDPBaseImplementation::createUDPSockets(){
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
//if eth is mistaken with ip address
@ -1015,16 +1107,11 @@ int UDPBaseImplementation::createNewFile(){
}
// This is actually called on CTRL-C
void UDPBaseImplementation::closeFile(int ithr)
{
void UDPBaseImplementation::closeFile(int ithr){
#ifdef VERBOSE
cout << "In closeFile for thread " << ithr << endl;
#endif
FILE_LOG(logDEBUG) << __AT__ << "called";
if(!dataCompression){
if(sfilefd){
@ -1072,6 +1159,9 @@ void UDPBaseImplementation::closeFile(int ithr){
#endif
}
FILE_LOG(logDEBUG) << __AT__ << "exited";
}

View File

@ -16,54 +16,112 @@
#include <sys/stat.h> // stat
#include <sys/socket.h> // socket(), bind(), listen(), accept(), shut down
#include <arpa/inet.h> // sock_addr_in, htonl, INADDR_ANY
#include <stdlib.h> // exit()
#include <iomanip> //set precision
#include <sys/mman.h> //munmap
#include <stdlib.h> // exit()
#include <iomanip> // set precision
#include <sys/mman.h> // munmap
#include <string.h>
#include <iostream>
#include <sstream>
//#include "utilities.h"
using namespace std;
/*
TODO
+ filePath != getFilePath
*/
UDPRESTImplementation::UDPRESTImplementation() : isInitialized(false), status(slsReceiverDefs::ERROR) {}
UDPRESTImplementation::UDPRESTImplementation(){
rest_hostname = "localhost";
rest_port = 8081;
}
UDPRESTImplementation::~UDPRESTImplementation(){}
void UDPRESTImplementation::initialize(const char *detectorHostName){
void UDPRESTImplementation::configure(map<string, string> config_map){
FILE_LOG(logWARNING) << __AT__ << " called";
string name;
if (detectorHostName != NULL)
name = detectorHostName;
map<string, string>::const_iterator pos;
if (name.empty()) {
FILE_LOG(logDEBUG) << "initialize(): can't initialize with empty string or NULL for detectorHostname";
} else if (isInitialized == true) {
FILE_LOG(logDEBUG) << "initialize(): already initialized, can't initialize several times";
} else {
FILE_LOG(logDEBUG) << "initialize(): initialize() with: detectorHostName=" << name;
strcpy(detHostname,detectorHostName);
//init_config.detectorHostname = name;
pos = config_map.find("rest_hostname");
if (pos != config_map.end() ){
string host_port_str = pos->second;
std::size_t pos = host_port_str.find(":"); // position of "live" in str
if(pos != string::npos){
istringstream (host_port_str.substr (pos)) >> rest_port;
rest_hostname = host_port_str.substr(0, pos);
cout << rest_hostname << " " << rest_port << endl;
}
}
//REST call - hardcoded
//RestHelper rest ;
rest->init(detHostname, 8080);
for(map<string, string>::const_iterator i=config_map.begin(); i != config_map.end(); i++){
std::cout << i->first << " " << i->second<< std::endl;
}
};
void UDPRESTImplementation::initialize_REST(){
if (rest_hostname.empty()) {
FILE_LOG(logDEBUG) << __AT__ <<"can't initialize with empty string or NULL for detectorHostname";
}
else if (isInitialized == true) {
FILE_LOG(logDEBUG) << __AT__ << "already initialized, can't initialize several times";
}
else {
FILE_LOG(logDEBUG) << __AT__ << "with receiverHostName=" << rest_hostname << ":" << rest_port;
rest = new RestHelper() ;
std::string answer;
int code = rest->get_json("status", &answer);
if (code != 0){
//throw -1;
std::cout << "I SHOULD THROW AN EXCEPTION!!!" << std::endl;
int code;
try{
rest->init(rest_hostname, rest_port);
code = rest->get_json("state", &answer);
if (code != 0){
throw answer;
}
else{
isInitialized = true;
status = slsReceiverDefs::IDLE;
}
FILE_LOG(logDEBUG) << __func__ << "Answer: " << answer;
}
else{
isInitialized = true;
status = slsReceiverDefs::IDLE;
catch(std::string e){
FILE_LOG(logERROR) << __func__ << ": " << e;
throw;
}
std::cout << "Answer: " << answer << std::endl;
//JsonBox::Object json_object;
//json_object["configfile"] = JsonBox::Value("FILENAME");
JsonBox::Value json_request;
//json_request["configfile"] = "config.py";
json_request["path"] = filePath;
stringstream ss;
string test;
std::cout << "GetSTring: " << json_request << std::endl;
json_request.writeToStream(ss, false);
//ss << json_request;
ss >> test;
cout << "aaaa" <<filePath << endl;
test = "{\"path\":\"" + string( getFilePath() ) + "\"}";
code = rest->post_json("state/initialize", &answer, test);
FILE_LOG(logDEBUG) << __AT__ << "state/configure got " << code;
code = rest->get_json("state", &answer);
FILE_LOG(logDEBUG) << __AT__ << "state got " << code << " " << answer;
/*
@ -73,14 +131,18 @@ void UDPRESTImplementation::initialize(const char *detectorHostName){
std::cout << "JSON " << json_value["status"] << std::endl;
*/
}
FILE_LOG(logDEBUG) << __func__ << ": initialize() done";
}
/*
int UDPRESTImplementation::setDetectorType(detectorType det){
cout << "[WARNING] This is a base implementation, " << __func__ << " not correctly implemented" << endl;
return OK;
}
*/
/*Frame indices and numbers caught*/
@ -180,7 +242,9 @@ char *UDPRESTImplementation::getDetectorHostname() const{
*/
void UDPRESTImplementation::setEthernetInterface(char* c){
strcpy(eth,c);
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
//strcpy(eth,c);
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " done";
}
@ -219,14 +283,17 @@ int32_t UDPRESTImplementation::setScanTag(int32_t stag){
*/
int32_t UDPRESTImplementation::setDynamicRange(int32_t dr){
cout << "Setting Dynamic Range" << endl;
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
int olddr = dynamicRange;
if(dr >= 0){
dynamicRange = dr;
}
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " " << getDynamicRange();
return getDynamicRange();
}
@ -571,6 +638,8 @@ void UDPRESTImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char
int UDPRESTImplementation::createUDPSockets(){
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
std::cout << "AAAAAAAAAAAa" << std::endl;
//if eth is mistaken with ip address
if (strchr(eth,'.')!=NULL)
@ -615,13 +684,24 @@ int UDPRESTImplementation::createUDPSockets(){
int UDPRESTImplementation::shutDownUDPSockets(){
FILE_LOG(logDEBUG) << __AT__ << "called";
FILE_LOG(logDEBUG) << __AT__ << "doing nothing";
/*
for(int i=0;i<numListeningThreads;i++){
if(udpSocket[i]){
FILE_LOG(logDEBUG) << __AT__ << " UDP socket " << i;
udpSocket[i]->ShutDownSocket();
delete udpSocket[i];
udpSocket[i] = NULL;
}
}
*/
FILE_LOG(logDEBUG) << __AT__ << "finished";
return OK;
}
@ -955,9 +1035,7 @@ int UDPRESTImplementation::createNewFile(){
void UDPRESTImplementation::closeFile(int ithr){
#ifdef VERBOSE
cout << "In closeFile for thread " << ithr << endl;
#endif
FILE_LOG(logDEBUG) << __AT__ << "called for thread " << ithr;
if(!dataCompression){
if(sfilefd){
@ -1005,6 +1083,8 @@ void UDPRESTImplementation::closeFile(int ithr){
#endif
}
FILE_LOG(logDEBUG) << __AT__ << "exited for thread " << ithr;
}
@ -1014,13 +1094,32 @@ void UDPRESTImplementation::closeFile(int ithr){
int UDPRESTImplementation::startReceiver(char message[]){
int i;
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
initialize_REST();
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " initialized";
// #ifdef VERBOSE
cout << "Starting Receiver" << endl;
//#endif
std::string answer;
int code;
//test = "{\"configfile\":\"config.pu\", \"path\":\"patto\"}";
code = rest->post_json("state/configure", &answer);
std::cout << answer << std::endl;
code = rest->get_json("state", &answer);
std::cout << answer << std::endl;
code = rest->post_json("state/open", &answer);
std::cout << answer << std::endl;
code = rest->get_json("state", &answer);
std::cout << answer << std::endl;
//reset listening thread variables
/*
measurementStarted = false;
//should be set to zero as its added to get next start frame indices for scans for eiger
if(!acqStarted) currframenum = 0;
@ -1028,8 +1127,9 @@ int UDPRESTImplementation::startReceiver(char message[]){
for(int i = 0; i < numListeningThreads; ++i)
totalListeningFrameCount[i] = 0;
*/
//udp socket
/*
if(createUDPSockets() == FAIL){
strcpy(message,"Could not create UDP Socket(s).\n");
cout << endl << message << endl;
@ -1037,7 +1137,8 @@ int UDPRESTImplementation::startReceiver(char message[]){
}
cout << "UDP socket(s) created successfully. 1st port " << server_port[0] << endl;
*/
/*
if(setupWriter() == FAIL){
//stop udp socket
shutDownUDPSockets();
@ -1069,7 +1170,7 @@ int UDPRESTImplementation::startReceiver(char message[]){
sem_post(&listensmp[i]);
for(i=0; i < numWriterThreads; ++i)
sem_post(&writersmp[i]);
*/
cout << "Receiver Started.\nStatus:" << status << endl;
@ -1081,10 +1182,7 @@ int UDPRESTImplementation::startReceiver(char message[]){
int UDPRESTImplementation::stopReceiver(){
//#ifdef VERBOSE
cout << "Stopping Receiver" << endl;
//#endif
FILE_LOG(logDEBUG) << __AT__ << "called";
if(status == RUNNING)
startReadout();
@ -1101,7 +1199,8 @@ int UDPRESTImplementation::stopReceiver(){
status = IDLE;
pthread_mutex_unlock(&(status_mutex));
cout << "Receiver Stopped.\nStatus:" << status << endl << endl;
FILE_LOG(logDEBUG) << __AT__ << "exited, status " << endl;
return OK;
}
@ -1110,10 +1209,7 @@ int UDPRESTImplementation::stopReceiver(){
void UDPRESTImplementation::startReadout(){
//#ifdef VERBOSE
cout << "Start Receiver Readout" << endl;
//#endif
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
//wait so that all packets which take time has arrived
usleep(50000);
@ -1127,6 +1223,7 @@ void UDPRESTImplementation::startReadout(){
//kill udp socket to tell the listening thread to push last packet
shutDownUDPSockets();
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " done";
}

View File

@ -127,7 +127,7 @@ void UDPStandardImplementation::initializeMembers(){
UDPStandardImplementation::UDPStandardImplementation(){
cout << "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAa" << endl;
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting" ;
thread_started = 0;
eth = NULL;
@ -963,7 +963,7 @@ int UDPStandardImplementation::shutDownUDPSockets(){
// TODO: add a destroyListeningThreads
int UDPStandardImplementation::createListeningThreads(bool destroy){
int i;
void* status;
@ -974,6 +974,8 @@ int UDPStandardImplementation::createListeningThreads(bool destroy){
listeningthreads_mask = 0x0;
pthread_mutex_unlock(&(status_mutex));
FILE_LOG(logDEBUG) << "Starting " << __func__ << endl;
if(!destroy){
//start listening threads

View File

@ -8,7 +8,7 @@
#include <sstream>
#include <fstream>
#include <stdlib.h>
#include <map>
#include <getopt.h>
#include "slsReceiver.h"
@ -30,10 +30,12 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
*/
//creating base receiver
map<string, string> configuration_map;
int tcpip_port_no = 1954;
success=OK;
string fname = "";
string udp_interface_type = "standard";
string rest_hostname = "localhost:8081";
//parse command line for config
static struct option long_options[] = {
@ -44,6 +46,7 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
{"type", required_argument, 0, 't'},
{"config", required_argument, 0, 'f'},
{"rx_tcpport", required_argument, 0, 'b'},
{"rest_hostname", required_argument, 0, 'r'},
{"help", no_argument, 0, 'h'},
{0, 0, 0, 0}
};
@ -52,7 +55,7 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
int c;
while ( c != -1 ){
c = getopt_long (argc, argv, "bfht", long_options, &option_index);
c = getopt_long (argc, argv, "bfhtr", long_options, &option_index);
/* Detect the end of the options. */
if (c == -1)
@ -72,12 +75,19 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
udp_interface_type = optarg;
break;
case 'r':
rest_hostname = optarg;
break;
case 'h':
string help_message = """\nSLS Receiver Server\n\n""";
help_message += """usage: slsReceiver --config config_fname [--rx_tcpport port]\n\n""";
help_message += """\t--config:\t configuration filename for SLS Detector receiver\n""";
help_message += """\t--rx_tcpport:\t TCP Communication Port with the client. Default: 1954.\n\n""";
help_message += """\t--rest_hostname:\t Receiver hostname:port. It applies only to REST receivers, and indicates the hostname of the REST backend. Default: localhost:8081.\n\n""";
help_message += """\t--type:\t Type of the receiver. Possible arguments are: standard, REST. Default: standard.\n\n""";
cout << help_message << endl;
break;
@ -91,7 +101,7 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
if( !fname.empty() ){
try{
FILE_LOG(logINFO) << "config file name " << fname;
success = read_config_file(fname, &tcpip_port_no);
success = read_config_file(fname, &tcpip_port_no, &configuration_map);
//VERBOSE_PRINT("Read configuration file of " + iline + " lines");
}
catch(...){
@ -108,6 +118,7 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
if (success==OK){
FILE_LOG(logINFO) << "SLS Receiver starting " << udp_interface_type << " on port " << tcpip_port_no << endl;
udp_interface = UDPInterface::create(udp_interface_type);
udp_interface->configure(configuration_map);
tcpipInterface = new slsReceiverTCPIPInterface(success, udp_interface, tcpip_port_no);
//tcp ip interface
}

View File

@ -69,14 +69,13 @@ slsReceiverTCPIPInterface::slsReceiverTCPIPInterface(int &success, UDPInterface*
//Catch signal SIGINT to close files properly
signal(SIGINT,staticCloseFile);
}
}
}
int slsReceiverTCPIPInterface::setPortNumber(int pn){
int p_number;
int p_number;
MySocketTCP *oldsocket=NULL;;
int sd=0;
@ -144,9 +143,11 @@ void slsReceiverTCPIPInterface::stop(){
cout<<"Shutting down TCP Socket and TCP thread"<<endl;
killTCPServerThread = 1;
socket->ShutDownSocket();
cout<<"Socket closed"<<endl;
void* status;
pthread_join(TCPServer_thread, &status);
killTCPServerThread = 0;
cout<<"Threads joined"<<endl;
}
@ -656,7 +657,7 @@ int slsReceiverTCPIPInterface::setup_udp(){
strcpy(eth,"");
ret = FAIL;
}
cout<<"eth:"<<eth<<endl;
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " " << eth;
receiverBase->setEthernetInterface(eth);
//get mac address from ethernet interface

View File

@ -5,19 +5,23 @@
#include <stdio.h>
#include <fstream>
#include <map>
#include "utilities.h"
using namespace std;
int read_config_file(string fname, int *tcpip_port_no){
int read_config_file(string fname, int *tcpip_port_no, map<string, string> * configuration_map ){
ifstream infile;
string sLine,sargname;
string sLine,sargname, sargvalue;
int iline = 0;
int success = slsReceiverDefs::OK;
infile.open(fname.c_str(), ios_base::in);
if (infile.is_open()) {
while(infile.good()){
@ -26,21 +30,25 @@ int read_config_file(string fname, int *tcpip_port_no){
//VERBOSE_PRINT(sLine);
if(sLine.find('#')!=string::npos){
//VERBOSE_PRINT( "Line is a comment ");
if(sLine.find('#') != string::npos)
continue;
}
else if(sLine.length()<2){
//VERBOSE_PRINT("Empty line ");
else if(sLine.length()<2)
continue;
}
else{
istringstream sstr(sLine);
//parameter name
if(sstr.good())
if(sstr.good()){
sstr >> sargname;
if (! sstr.good())
continue;
sstr >> sargvalue;
(*configuration_map)[sargname] = sargvalue;
}
//tcp port
if(sargname=="rx_tcpport"){
if(sstr.good()) {