PGM2 sender

5 files changed, 100 insertions, 24 deletions

diff --git a/src/pgm_receiver.cpp b/src/pgm_receiver.cpp
index 50a8ff9..7af6ed5 100644
--- a/src/pgm_receiver.cpp
+++ b/src/pgm_receiver.cpp
@@ -153,7 +153,6 @@ void zmq::pgm_receiver_t::in_event ()
         // information (sizeof uint16_t).
         raw_data +=  sizeof (uint16_t);
         nbytes -= sizeof (uint16_t);
-        zmq_assert (apdu_offset <= nbytes);
 
         //  New peer.
         if (it == peers.end ()) {
@@ -174,6 +173,7 @@ void zmq::pgm_receiver_t::in_event ()
         //  Now is the possibility to join the stream.
         if (!it->second.joined) {
  
+            zmq_assert (apdu_offset <= nbytes);
             zmq_assert (it->second.decoder == NULL);
 
             //  We have to move data to the begining of the first message.
diff --git a/src/pgm_sender.cpp b/src/pgm_sender.cpp
index 423865b..51dfbf1 100644
--- a/src/pgm_sender.cpp
+++ b/src/pgm_sender.cpp
@@ -68,13 +68,21 @@ void zmq::pgm_sender_t::plug (i_inout *inout_)
 {
     
     //  Alocate 2 fds for PGM socket.
-    int downlink_socket_fd;
-    int uplink_socket_fd;
+    int downlink_socket_fd = 0;
+    int uplink_socket_fd = 0;
+#ifdef ZMQ_HAVE_OPENPGM2
+    int rdata_notify_fd = 0;
+#endif
 
     encoder.set_inout (inout_);
 
     //  Fill fds from PGM transport.
+#ifdef ZMQ_HAVE_OPENPGM1
     pgm_socket.get_sender_fds (&downlink_socket_fd, &uplink_socket_fd);
+#elif ZMQ_HAVE_OPENPGM2
+    pgm_socket.get_sender_fds 
+        (&downlink_socket_fd, &uplink_socket_fd, &rdata_notify_fd);
+#endif
 
     //  Add downlink_socket_fd into poller.
     handle = add_fd (downlink_socket_fd);
@@ -82,9 +90,17 @@ void zmq::pgm_sender_t::plug (i_inout *inout_)
     //  Add uplink_socket_fd into the poller.
     uplink_handle = add_fd (uplink_socket_fd);
 
+    //  Add rdata_notify_fd into the poller.
+#ifdef ZMQ_HAVE_OPENPGM2
+    rdata_notify_handle = add_fd (rdata_notify_fd);   
+#endif
+
     //  Set POLLIN. We wont never want to stop polling for uplink = we never
     //  want to stop porocess NAKs.
     set_pollin (uplink_handle);
+#ifdef ZMQ_HAVE_OPENPGM2
+    set_pollin (rdata_notify_handle);
+#endif
 
     //  Set POLLOUT for downlink_socket_handle.
     set_pollout (handle);
@@ -96,6 +112,9 @@ void zmq::pgm_sender_t::unplug ()
 {
     rm_fd (handle);
     rm_fd (uplink_handle);
+#ifdef ZMQ_HAVE_OPENPGM2
+    rm_fd (rdata_notify_handle);
+#endif
     encoder.set_inout (NULL);
     inout = NULL;
 }
@@ -167,11 +186,12 @@ void zmq::pgm_sender_t::out_event ()
             zmq_log (1, "pgm rate limit reached, %s(%i)\n", __FILE__, __LINE__);
         }
 
+#ifdef ZMQ_HAVE_OPENPGM1
         //  After sending data slice is owned by tx window.
         if (nbytes) {
             out_buffer = NULL;
         }
-
+#endif
         write_pos += nbytes;
     }
 
diff --git a/src/pgm_sender.hpp b/src/pgm_sender.hpp
index 8fdda6c..380e80b 100644
--- a/src/pgm_sender.hpp
+++ b/src/pgm_sender.hpp
@@ -76,6 +76,9 @@ namespace zmq
         //  Poll handle associated with PGM socket.
         handle_t handle;
         handle_t uplink_handle;
+#ifdef ZMQ_HAVE_OPENPGM2
+        handle_t rdata_notify_handle;
+#endif
 
         //  Parent session.
         i_inout *inout;
diff --git a/src/pgm_socket.cpp b/src/pgm_socket.cpp
index 65a80a5..dbfe52d 100644
--- a/src/pgm_socket.cpp
+++ b/src/pgm_socket.cpp
@@ -39,7 +39,7 @@
 #include "uuid.hpp"
 
 //#define PGM_SOCKET_DEBUG
-//#define PGM_SOCKET_DEBUG_LEVEL 1
+//#define PGM_SOCKET_DEBUG_LEVEL 4
 
 // level 1 = key behaviour
 // level 2 = processing flow
@@ -275,7 +275,7 @@ int zmq::pgm_socket_t::open_transport (void)
         
         //  Set transport->can_send_data = FALSE.
         //  Note that NAKs are still generated by the transport.
-        rc = pgm_transport_set_recv_only (g_transport, false);
+        rc = pgm_transport_set_recv_only (g_transport, true, false);
         if (rc != pgm_ok) {
             errno = EINVAL;
             return -1;
@@ -511,8 +511,18 @@ int zmq::pgm_socket_t::get_receiver_fds (int *recv_fd_,
 //   Get fds and store them into user allocated memory. 
 //   sender_fd is from pgm_transport->send_sock.
 //   receive_fd_ is from  transport->recv_sock.
-int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_)
+int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_, 
+    int *rdata_notify_fd_)
 {
+#ifdef ZMQ_HAVE_OPENPGM1
+    zmq_assert (send_fd_);
+    zmq_assert (receive_fd_);
+    zmq_assert (!rdata_notify_fd_);
+#elif ZMQ_HAVE_OPENPGM2
+    zmq_assert (send_fd_);
+    zmq_assert (receive_fd_);
+    zmq_assert (rdata_notify_fd_);
+#endif
 
     //  Preallocate pollfds array.
     int fds_array_size = pgm_sender_fd_count;
@@ -530,8 +540,14 @@ int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_)
     zmq_assert (rc == pgm_sender_fd_count);
  
     //  Store pfds into user allocated space.
+#ifdef ZMQ_HAVE_OPENPGM1
     *receive_fd_ = fds [0].fd;
     *send_fd_ = fds [1].fd;
+#elif ZMQ_HAVE_OPENPGM2
+    *receive_fd_ = fds [0].fd;
+    *rdata_notify_fd_ = fds [1].fd;
+    *send_fd_ = fds [2].fd;
+#endif
 
     delete [] fds;
 
@@ -542,10 +558,9 @@ int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_)
 size_t zmq::pgm_socket_t::send (unsigned char *data_, size_t data_len_)
 {
 
-    ssize_t nbytes = 0;
-
 #ifdef ZMQ_HAVE_OPENPGM1
 
+    ssize_t nbytes = 0;
     iovec iov = {data_,data_len_};
 
     nbytes = pgm_transport_send_packetv (g_transport, &iov, 1, 
@@ -561,13 +576,30 @@ size_t zmq::pgm_socket_t::send (unsigned char *data_, size_t data_len_)
     //  now. We have to call write_one_pkt again.
     nbytes = nbytes == -1 ? 0 : nbytes;
 
-    zmq_log (4, "wrote %iB, %s(%i)\n", (int)nbytes, __FILE__, __LINE__);
+#elif ZMQ_HAVE_OPENPGM2
+
+    size_t nbytes = 0;
+   
+    PGMIOStatus status = pgm_send (g_transport, data_, data_len_, &nbytes);
+
+    if (nbytes != data_len_) {
+        zmq_log (1, "status %i, data_len %i, wrote %iB, %s(%i)\n", 
+            (int) status, (int) data_len_, (int) nbytes, __FILE__, __LINE__);
+
+        zmq_assert (status == PGM_IO_STATUS_AGAIN2);
+        zmq_assert (nbytes == 0);
+    }
+#endif
+
+    zmq_log (4, "wrote %i/%iB, %s(%i)\n", (int) nbytes, (int) data_len_,
+        __FILE__, __LINE__);
     
     // We have to write all data as one packet.
     if (nbytes > 0) {
-        zmq_assert (nbytes == (ssize_t)data_len_);
+        zmq_log (1, "data sent %i, %s(%i)\n", (int) nbytes, 
+            __FILE__, __LINE__);
+        zmq_assert ((ssize_t) nbytes == (ssize_t) data_len_);
     }
-#endif
 
     return nbytes;
 }
@@ -603,15 +635,17 @@ size_t zmq::pgm_socket_t::get_max_apdu_at_once (size_t readbuf_size_)
 //  content via pgm_transport_send() calls or unused with pgm_packetv_free1(). 
 void *zmq::pgm_socket_t::get_buffer (size_t *size_)
 {
-#ifdef ZMQ_HAVE_OPENPGM1
     //  Store size.
     *size_ = get_max_tsdu_size ();
 
-    //  Allocate one packet.
+#ifdef ZMQ_HAVE_OPENPGM1
+    //  Allocate one packet in tx window.
     return pgm_packetv_alloc (g_transport, false);
 #elif ZMQ_HAVE_OPENPGM2
-    zmq_assert (false);
-
+    //  Allocate buffer.
+    unsigned char *apdu_buff = new unsigned char [*size_]; 
+    zmq_assert (apdu_buff);
+    return apdu_buff;
 #endif
 }
 
@@ -622,7 +656,7 @@ void zmq::pgm_socket_t::free_buffer (void *data_)
 #ifdef ZMQ_HAVE_OPENPGM1
     pgm_packetv_free1 (g_transport, data_, false);
 #elif ZMQ_HAVE_OPENPGM2
-    zmq_assert (false);
+    delete [] (unsigned char*) data_;
 #endif
 }
 
@@ -718,6 +752,8 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_)
             zmq_log (1, "PGMIOStatus %i, nbytes_rec %i, %s(%i).\n", 
                 status, (int) nbytes_rec, __FILE__, __LINE__);
 
+            zmq_assert (false);
+
             nbytes_rec = 0;
             return -1;
         }
@@ -767,21 +803,34 @@ void zmq::pgm_socket_t::process_upstream (void)
 {
     zmq_log (1, "On upstream packet, %s(%i)\n", __FILE__, __LINE__);
 
-    ssize_t dummy_bytes = 0;
+    pgm_msgv_t dummy_msg;
 
 #ifdef ZMQ_HAVE_OPENPGM1
+    ssize_t dummy_bytes = 0;
+
     //  We acctually do not want to read any data here we are going to 
     //  process NAK.
-    pgm_msgv_t dummy_msg;
 
     dummy_bytes = pgm_transport_recvmsgv (g_transport, &dummy_msg,
         1, MSG_DONTWAIT);
+
+    //  No data should be returned.
+    zmq_assert (dummy_bytes == -1 && errno == EAGAIN);
+
 #elif defined ZMQ_HAVE_OPENPGM2
-    zmq_assert (false);
+    size_t dummy_bytes = 0;
+    GError *pgm_error = NULL;
+
+    PGMIOStatus status = pgm_recvmsgv (g_transport, &dummy_msg,
+        1, MSG_DONTWAIT, &dummy_bytes, &pgm_error);
+
+    zmq_log (1, "upstream status %i, nbytes %i, %s(%i)\n",
+        (int) status, (int) dummy_bytes, __FILE__, __LINE__);
+
+    //  No data should be returned.
+    zmq_assert (dummy_bytes == 0 && status == PGM_IO_STATUS_AGAIN);
 #endif
     
-    //  No data should be returned.
-    zmq_assert (dummy_bytes == -1 && errno == EAGAIN);
 }
 
 #endif
diff --git a/src/pgm_socket.hpp b/src/pgm_socket.hpp
index 5225e50..473e58c 100644
--- a/src/pgm_socket.hpp
+++ b/src/pgm_socket.hpp
@@ -65,7 +65,7 @@ namespace zmq
 
         //   Get sender and receiver fds and store it to user allocated 
         //   memory. Receive fd is used to process NAKs from peers.
-        int get_sender_fds (int *send_fd_, int *receive_fd_);
+        int get_sender_fds (int *send_fd_, int *receive_fd_, int *rdata_notify_fd_ = NULL);
 
         //  Send data as one APDU, transmit window owned memory.
         size_t send (unsigned char *data_, size_t data_len_);
@@ -143,8 +143,12 @@ namespace zmq
         size_t pgm_msgv_len;
 
         //  Sender transport uses 2 fd.
+#ifdef ZMQ_HAVE_OPENPGM1
         enum {pgm_sender_fd_count = 2};
-    
+#elif ZMQ_HAVE_OPENPGM2
+        enum {pgm_sender_fd_count = 3};
+#endif
+
         //  Receiver transport uses 2 fd.
         enum {pgm_receiver_fd_count = 2};
 #endif