Migrate to new NetworkChannel interface

CMakeLists.txt

@@ -30,6 +30,7 @@ endif()
 if(BUILD_TESTS)
   set(BUILD_LF_TESTS ON)
   set(BUILD_UNIT_TESTS ON)
+  set(NETWORK_POSIX_TCP ON) # TODO: This is currently needed because one of the tests uses this stack, we need a nicer way of selecting build options for tests and apps.
   set(CMAKE_BUILD_TYPE "Debug")
   find_program(CLANG_TIDY clang-tidy)
   if (CLANG_TIDY)

examples/posix/testing_fed_conn_receiver.c

@@ -63,14 +63,15 @@ typedef struct {
 
 void RecvSenderBundle_ctor(RecvSenderBundle *self, Reactor *parent) {
   ConnRecv_ctor(&self->conn, parent);
-  TcpIpChannel_ctor(&self->channel, "127.0.0.1", PORT_NUM, AF_INET);
+  TcpIpChannel_ctor(&self->channel, "127.0.0.1", PORT_NUM, AF_INET, false);
   self->inputs[0] = &self->conn.super;
 
   NetworkChannel *channel = (NetworkChannel *)&self->channel;
+  channel->open_connection(channel);
 
   lf_ret_t ret;
   do {
-    ret = channel->connect(channel);
+    ret = channel->try_connect(channel);
   } while (ret != LF_OK);
   validate(ret == LF_OK);
   printf("Recv: Connected\n");

examples/posix/testing_fed_conn_sender.c

@@ -106,12 +106,12 @@ typedef struct {
 } SenderRecvBundle;
 
 void SenderRecvConn_ctor(SenderRecvBundle *self, Sender *parent) {
-  TcpIpChannel_ctor(&self->channel, "127.0.0.1", PORT_NUM, AF_INET);
+  TcpIpChannel_ctor(&self->channel, "127.0.0.1", PORT_NUM, AF_INET, true);
   ConnSender_ctor(&self->conn, &parent->super, &self->super);
   self->output[0] = &self->conn.super;
 
   NetworkChannel *channel = (NetworkChannel *)&self->channel;
-  int ret = channel->bind(channel);
+  lf_ret_t ret = channel->open_connection(channel);
   if (ret != LF_OK) {
     printf("bind failed with %d\n", errno);
     exit(1);
@@ -120,8 +120,10 @@ void SenderRecvConn_ctor(SenderRecvBundle *self, Sender *parent) {
   printf("Sender: Bound\n");
 
   // accept one connection
-  bool new_connection = channel->accept(channel);
-  validate(new_connection);
+  do {
+    ret = channel->try_connect(channel);
+  } while (ret != LF_OK);
+  validate(ret == LF_OK);
   printf("Sender: Accepted\n");
 
   self->serialize_hooks[0] = serialize_msg_t;

examples/riot/blinky/Makefile

@@ -13,6 +13,6 @@ DEVELHELP ?= 1
 QUIET ?= 1
 
 # Enable reactor-uc features
-# CFLAGS += -DNETWORK_POSIX_TCP
+CFLAGS += -DNETWORK_POSIX_TCP
 
 include $(CURDIR)/../../../make/riot/riot.mk

examples/zephyr/basic_federated/federated_receiver1/src/receiver.c

@@ -61,15 +61,16 @@ typedef struct {
 
 void RecvSenderBundle_ctor(RecvSenderBundle *self, Reactor *parent) {
   ConnRecv_ctor(&self->conn, parent);
-  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_NUM, AF_INET);
+  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_NUM, AF_INET, false);
   self->inputs[0] = &self->conn.super;
 
   NetworkChannel *chan = &self->chan.super;
+  chan->open_connection(channel);
 
   lf_ret_t ret;
   LF_DEBUG(ENV, "Recv: Connecting");
   do {
-    ret = chan->connect(chan);
+    ret = chan->try_connect(chan);
   } while (ret != LF_OK);
   LF_DEBUG(ENV, "Recv: Connected");
 

examples/zephyr/basic_federated/federated_receiver2/src/receiver.c

@@ -61,15 +61,16 @@ typedef struct {
 
 void RecvSenderBundle_ctor(RecvSenderBundle *self, Reactor *parent) {
   ConnRecv_ctor(&self->conn, parent);
-  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_NUM, AF_INET);
+  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_NUM, AF_INET, false);
   self->inputs[0] = &self->conn.super;
 
   NetworkChannel *chan = &self->chan.super;
+  chan->open_connection(chan);
 
   lf_ret_t ret;
   LF_DEBUG(ENV, "Recv: Connecting");
   do {
-    ret = chan->connect(chan);
+    ret = chan->try_connect(chan);
   } while (ret != LF_OK);
   LF_DEBUG(ENV, "Recv: Connected");
 

examples/zephyr/basic_federated/federated_sender/src/sender.c

@@ -115,37 +115,41 @@ typedef struct {
 } SenderRecv2Bundle;
 
 void SenderRecv1Bundle_ctor(SenderRecv1Bundle *self, Reactor *parent) {
-  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_CONN_1, AF_INET);
+  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_CONN_1, AF_INET, true);
   ConnSender1_ctor(&self->conn, parent, &self->super);
   self->output[0] = &self->conn.super;
 
   NetworkChannel *chan = &self->chan.super;
-  int ret = chan->bind(chan);
+  lf_ret_t ret = chan->open_connection(chan);
   validate(ret == LF_OK);
   printf("Sender: Bound 1\n");
 
   // accept one connection
-  bool new_connection = chan->accept(chan);
-  validate(new_connection);
+  do {
+    ret = chan->try_connect(chan);
+  } while (ret != LF_OK);
+  validate(ret == LF_OK);
   printf("Sender: Accepted 1\n");
 
   FederatedConnectionBundle_ctor(&self->super, parent, &self->chan.super, NULL, 0,
                                  (FederatedOutputConnection **)&self->output, 1);
 }
 
 void SenderRecv2Bundle_ctor(SenderRecv2Bundle *self, Reactor *parent) {
-  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_CONN_2, AF_INET);
+  TcpIpChannel_ctor(&self->chan, "192.168.1.100", PORT_CONN_2, AF_INET, true);
   ConnSender2_ctor(&self->conn, parent, &self->super);
   self->output[0] = &self->conn.super;
 
   NetworkChannel *chan = &self->chan.super;
-  int ret = chan->bind(chan);
+  lf_ret_t ret = chan->open_connection(chan);
   validate(ret == LF_OK);
   printf("Sender: Bound 2\n");
 
   // accept one connection
-  bool new_connection = chan->accept(chan);
-  validate(new_connection);
+  do {
+    ret = chan->try_connect(chan);
+  } while (ret != LF_OK);
+  validate(ret == LF_OK);
   printf("Sender: Accepted 2\n");
 
   FederatedConnectionBundle_ctor(&self->super, parent, &self->chan.super, NULL, 0,

include/reactor-uc/network_channel.h

@@ -13,15 +13,14 @@ typedef struct FederatedConnectionBundle FederatedConnectionBundle;
 typedef struct NetworkChannel NetworkChannel;
 
 struct NetworkChannel {
-  lf_ret_t (*bind)(NetworkChannel *self);
-  lf_ret_t (*connect)(NetworkChannel *self);
-  bool (*accept)(NetworkChannel *self);
-  void (*close)(NetworkChannel *self);
-  void (*register_callback)(NetworkChannel *self,
-                            void (*receive_callback)(FederatedConnectionBundle *conn, TaggedMessage *message),
-                            FederatedConnectionBundle *conn);
-  lf_ret_t (*send)(NetworkChannel *self, TaggedMessage *message);
-  TaggedMessage *(*receive)(NetworkChannel *self);
+  size_t dest_channel_id; // So that we can "address" one of several NetworkChannel's at the other end.
+  lf_ret_t (*open_connection)(NetworkChannel *self);
+  lf_ret_t (*try_connect)(NetworkChannel *self);
+  void (*close_connection)(NetworkChannel *self);
+  lf_ret_t (*send_blocking)(NetworkChannel *self, TaggedMessage *message);
+  void (*register_receive_callback)(NetworkChannel *self,
+                                    void (*receive_callback)(FederatedConnectionBundle *conn, TaggedMessage *message),
+                                    FederatedConnectionBundle *conn);
   void (*free)(NetworkChannel *self);
 };
 

include/reactor-uc/platform/posix/tcp_ip_channel.h

@@ -33,7 +33,6 @@ struct TcpIpChannel {
 
   fd_set set;
   bool server;
-  bool blocking;
   bool terminate;
 
   // required for callbacks
@@ -45,8 +44,6 @@ struct TcpIpChannel {
   void (*receive_callback)(FederatedConnectionBundle *conn, TaggedMessage *message);
 };
 
-void TcpIpChannel_ctor(TcpIpChannel *self, const char *host, unsigned short port, int protocol_family);
-
-void TcpIpChannel_free(NetworkChannel *self);
+void TcpIpChannel_ctor(TcpIpChannel *self, const char *host, unsigned short port, int protocol_family, bool server);
 
 #endif

src/federated.c

@@ -57,7 +57,7 @@ void FederatedOutputConnection_cleanup(Trigger *trigger) {
 
   LF_DEBUG(FED, "FedOutConn %p sending message with tag=%" PRId64 ":%" PRIu32, trigger, msg.tag.time,
            msg.tag.microstep);
-  lf_ret_t ret = channel->send(channel, &msg);
+  lf_ret_t ret = channel->send_blocking(channel, &msg);
 
   if (ret != LF_OK) {
     LF_ERR(FED, "FedOutConn %p failed to send message", trigger);
@@ -204,5 +204,5 @@ void FederatedConnectionBundle_ctor(FederatedConnectionBundle *self, Reactor *pa
   self->serialize_hooks = serialize_hooks;
 
   // Register callback function for message received.
-  self->net_channel->register_callback(self->net_channel, FederatedConnectionBundle_msg_received_cb, self);
+  self->net_channel->register_receive_callback(self->net_channel, FederatedConnectionBundle_msg_received_cb, self);
 }