zeromq stuff running

Makefile

@@ -1,10 +1,11 @@
 CC = gcc
-CFLAGS =  -std=c99 -I ./ -I ./include -DUA_NO_AMALGAMATION        # put pre-processor settings (-I, -D, etc) here
+CXX = g++
+CFLAGS =  -std=c99 -I ./ -I ./include -I /opt/zeromq/include -DUA_NO_AMALGAMATION        # put pre-processor settings (-I, -D, etc) here
 LDFLAGS =         # put linker settings here
 LIBS = ./lib/libopen62541-static.a
 
 .PHONY: all
-all:	myServer server_method server_folders myTestModelServer
+all:	myServer server_method server_folders myMqttTestModelServer myZeromqTestModelServer myZeromqTemperatureServer myZeromqTemperatureSniffer
 
 myServer: myServer.o 
 	$(CC) -o $@ $(CFLAGS) $(LDFLAGS) $^ $(LIBS)
@@ -20,15 +21,26 @@ TestModel.c:	testmodel.xml
 
 TestModel.h:	TestModel.c
 
-myTestModelServer.o:	myTestModelServer.c TestModel.h
+myMqttTestModelServer.o:	myMqttTestModelServer.c TestModel.h
 
-myTestModelServer: myTestModelServer.o TestModel.o 
+myMqttTestModelServer: myMqttTestModelServer.o TestModel.o 
 	$(CC) -o $@ $(CFLAGS) $(LDFLAGS) $^ $(LIBS) -lmosquitto
 
+myZeromqTestModelServer.o:	myZeromqTestModelServer.c TestModel.h
+
+myZeromqTestModelServer: myZeromqTestModelServer.o TestModel.o 
+	$(CXX) -o $@ $(CFLAGS) $(LDFLAGS) $^ $(LIBS) /opt/zeromq/lib/libzmq-static.a -lpthread
+
+myZeromqTemperatureServer: myZeromqTemperatureServer.o
+	$(CXX) -o $@ $(CFLAGS) $(LDFLAGS) $^ $(LIBS) /opt/zeromq/lib/libzmq-static.a -lpthread
+
+myZeromqTemperatureSniffer: myZeromqTemperatureSniffer.o
+	$(CXX) -o $@ $(CFLAGS) $(LDFLAGS) $^ $(LIBS) /opt/zeromq/lib/libzmq-static.a -lpthread
+
 .c.o:
 	$(CC) $(CFLAGS) -c $<
 
 .PHONY: clean
 clean:
-	-rm -f *.o myServer server_method server_folders myTestModelServer
+	-rm -f *.o myServer server_method server_folders myMqttTestModelServer myZeromqTestModelServer myZeromqTemperatureServer myZeromqTemperatureSniffer
 

myZeromqTemperatureServer.c

@@ -0,0 +1,43 @@
+#include <stdlib.h>
+#include <string.h>
+#include <zmq.h>
+#include <unistd.h>
+#include <signal.h>
+
+int running = 1;
+static void stopHandler(int sign) {
+  printf("received ctrl-c\n");
+  running = 0;
+}
+
+
+
+int main() {
+  signal(SIGINT, stopHandler); /* catches ctrl-c */
+
+  void *context = zmq_ctx_new();
+  void *publisher = zmq_socket(context, ZMQ_PUB);
+  int rc = zmq_bind(publisher, "tcp://127.0.0.1:12345");
+  if (rc != 0) {
+    printf("something wrong with zmq_bind: %s\n", zmq_strerror(errno));
+    exit(1);
+  }
+
+  double temperature = 15.0;
+  while (running) {
+    rc = zmq_send(publisher, &temperature, sizeof(double), 0);
+    if (rc != -1) {
+      printf("sent %f\n", temperature);
+    } else {
+      printf("something wrong with zmq_send: %s\n", zmq_strerror(errno));
+    }
+    
+    temperature += 0.25;
+
+    sleep(5);
+  }
+
+  printf("Terminating ...\n");
+  zmq_close(publisher);
+  zmq_ctx_destroy(context);
+}

myZeromqTemperatureSniffer.c

@@ -0,0 +1,43 @@
+#include <stdlib.h>
+#include <assert.h>
+#include <signal.h>
+#include <zmq.h>
+
+double measuredTemperature = 0.0;
+
+
+int running = 1;
+static void stopHandler(int sign) {
+  printf("received ctrl-c\n");
+  running = 0;
+}
+
+
+
+int main(int argc, char** argv) {
+  signal(SIGINT, stopHandler); /* catches ctrl-c */
+
+  /* prepare zmq context, connect to server and subscribe with empty filter */
+  void *zmqContext = zmq_ctx_new();
+  void *zmqSubscriber = zmq_socket(zmqContext, ZMQ_SUB);
+  int rc = zmq_connect(zmqSubscriber, "tcp://127.0.0.1:12345");
+  assert(rc == 0);
+  rc = zmq_setsockopt(zmqSubscriber, ZMQ_SUBSCRIBE, 0, 0);
+  assert(rc == 0);
+
+  while (running) {
+    double oldMeasuredTemperature = measuredTemperature;
+    
+    rc = zmq_recv(zmqSubscriber, &measuredTemperature, 
+		  sizeof(double), 0);
+    assert(rc != -1);
+      
+    if (oldMeasuredTemperature != measuredTemperature) {
+      printf("temperature value changed: %f\n", measuredTemperature);
+    }
+  }
+
+  printf("Terminating ...\n");
+  zmq_close(zmqSubscriber);
+  zmq_ctx_destroy(zmqContext);
+}

myZeromqTestModelServer.c

@@ -0,0 +1,146 @@
+#include <signal.h>
+#include <stdlib.h>
+#include <time.h>
+#include <sys/select.h>
+
+#include <ua_types.h>
+#include <ua_server.h>
+#include <logger_stdout.h>
+#include <networklayer_tcp.h>
+#include <ua_config_standard.h>
+
+#include <zmq.h>
+
+#include "TestModel.h"
+
+/*
+ * shared data
+ */
+uint32_t upTime = 0;
+double measuredTemperature = 0;
+
+
+
+
+UA_Logger logger = Logger_Stdout;
+UA_Boolean running = true;
+
+
+
+static void stopHandler(int sign) {
+  UA_LOG_INFO(logger, UA_LOGCATEGORY_SERVER, "received ctrl-c");
+  running = false;
+}
+
+static UA_StatusCode
+readInteger(void *handle, const UA_NodeId nodeid, UA_Boolean sourceTimeStamp,
+            const UA_NumericRange *range, UA_DataValue *dataValue) {
+  dataValue->hasValue = true;
+  UA_Variant_setScalarCopy(&dataValue->value, handle, &UA_TYPES[UA_TYPES_INT32]);
+  return UA_STATUSCODE_GOOD;
+}
+
+static UA_StatusCode
+readDouble(void *handle, const UA_NodeId nodeid, UA_Boolean sourceTimeStamp,
+	   const UA_NumericRange *range, UA_DataValue *dataValue) {
+  dataValue->hasValue = true;
+  UA_Variant_setScalarCopy(&dataValue->value, handle, &UA_TYPES[UA_TYPES_DOUBLE]);
+  return UA_STATUSCODE_GOOD;
+}
+
+
+uint64_t millis() {
+  struct timeval te;
+  gettimeofday(&te, NULL);
+  uint64_t ms = te.tv_sec * 1000 + te.tv_usec / 1000;
+  return ms;
+}
+
+
+int main(int argc, char** argv) {
+  signal(SIGINT, stopHandler); /* catches ctrl-c */
+
+  /* initialize the OPC-UA server */
+  UA_ServerConfig config = UA_ServerConfig_standard;
+  UA_ServerNetworkLayer nl = UA_ServerNetworkLayerTCP(UA_ConnectionConfig_standard, 16664);
+  config.networkLayers = &nl;
+  config.networkLayersSize = 1;
+  UA_Server *server = UA_Server_new(config);
+
+  /* create nodes from nodeset */
+  TestModel(server);
+
+  /* assign data sources to OPC-UA variables */
+  UA_DataSource uptimeDataSource = (UA_DataSource) {
+    .handle = &upTime, .read = readInteger, .write = 0};
+  UA_Server_setVariableNode_dataSource(server, UA_NODEID_NUMERIC(2, 6006),
+				       uptimeDataSource);
+
+
+  UA_DataSource measuredTemperatureDataSource = (UA_DataSource) {
+    .handle = &measuredTemperature, .read = readDouble, .write = 0};
+  UA_Server_setVariableNode_dataSource(server, UA_NODEID_NUMERIC(2, 6004),
+				       measuredTemperatureDataSource);
+
+  /* prepare OPC-UA server to run cooperative in a loop (below) */
+  UA_StatusCode retval = UA_Server_run_startup(server);
+  assert(retval == UA_STATUSCODE_GOOD);
+
+
+  /* prepare zmq context, connect to server and subscribe with empty filter */
+  void *zmqContext = zmq_ctx_new();
+  void *zmqSubscriber = zmq_socket(zmqContext, ZMQ_SUB);
+  int rc = zmq_connect(zmqSubscriber, "tcp://127.0.0.1:12345");
+  assert(rc == 0);
+  rc = zmq_setsockopt(zmqSubscriber, ZMQ_SUBSCRIBE, 0, 0);
+  assert(rc == 0);
+
+     
+  while (running == true) {
+    upTime = time(0);
+    // measuredTemperature += 0.25;
+	
+    uint16_t canWait = 0;
+      
+    canWait = UA_Server_run_iterate(server, 0);
+    // UA_LOG_INFO(logger, UA_LOGCATEGORY_USERLAND, "canWait: %i", canWait);
+
+    uint64_t beforeZmqRecv = millis();
+    double oldMeasuredTemperature = measuredTemperature;
+
+    /* we know we will receive a double, write it directly into the shared variable */
+    int rcLen = zmq_recv(zmqSubscriber, &measuredTemperature, 
+			 sizeof(double), ZMQ_DONTWAIT);
+    if (rcLen == -1) {
+      if (errno != EAGAIN) {
+	UA_LOG_INFO(logger, UA_LOGCATEGORY_USERLAND, "something wrong with zmq_recv: %i", errno);
+	exit(1);
+      }
+    } else {
+      UA_LOG_INFO(logger, UA_LOGCATEGORY_USERLAND, "something received: %i", rcLen);
+    }
+      
+    if (oldMeasuredTemperature != measuredTemperature) {
+      UA_LOG_INFO(logger, UA_LOGCATEGORY_USERLAND, "temperature value changed: %f", measuredTemperature);
+    }
+
+    /* measure duration of the zmq handling */
+    uint64_t durationZmqRecv = millis() - beforeZmqRecv;
+    canWait = (canWait > durationZmqRecv) ? (canWait - durationZmqRecv) : 0;
+    UA_LOG_INFO(logger, UA_LOGCATEGORY_USERLAND, "canWait: %i", canWait);
+
+
+    /* cooperative wait */  
+    struct timeval timeout = { .tv_sec = 0, .tv_usec = canWait * 1000 };
+    select(0, 0, 0, 0, &timeout);
+  }
+
+    
+  /* ctrl-c received -> clean up */
+  UA_Server_delete(server);
+  nl.deleteMembers(&nl);
+
+  zmq_close(zmqSubscriber);
+  zmq_ctx_destroy(zmqContext);
+  // return (int)retval;
+}