link libzmq with glib when congifured --with-pgm

5 files changed, 79 insertions, 15 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 68b34fa..f4f338e 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -62,6 +62,8 @@ libzmq_la_SOURCES = $(pgm_sources) \
     object.hpp \
     options.hpp \
     owned.hpp \
+    pgm_sender.hpp \
+    pgm_socket.hpp \
     pipe.hpp \
     platform.hpp \
     poll.hpp \
@@ -101,6 +103,8 @@ libzmq_la_SOURCES = $(pgm_sources) \
     object.cpp \
     options.cpp \
     owned.cpp \
+    pgm_sender.cpp \
+    pgm_socket.cpp \
     pipe.cpp \
     poll.cpp \
     select.cpp \
@@ -122,7 +126,7 @@ libzmq_la_SOURCES = $(pgm_sources) \
     zmq_listener.cpp \
     zmq_listener_init.cpp
 
-libzmq_la_LDFLAGS = -version-info @LTVER@
+libzmq_la_LDFLAGS = -version-info @LTVER@ @LIBZMQ_EXTRA_LDFLAFS@
 
 if BUILD_PGM
 libzmq_la_CXXFLAGS = -I$(top_srcdir)/foreign/openpgm/@pgm_basename@/openpgm/pgm/include/ -Wall @LIBZMQ_EXTRA_CXXFLAGS@
diff --git a/src/config.hpp b/src/config.hpp
index 17e67b9..43a4513 100644
--- a/src/config.hpp
+++ b/src/config.hpp
@@ -70,8 +70,10 @@ namespace zmq
 
         //  Maximal number of non-accepted connections that can be held by
         //  TCP listener object.
-        tcp_connection_backlog = 10
+        tcp_connection_backlog = 10,
 
+        //  Maximum transport data unit size for PGM (TPDU).
+        pgm_max_tpdu = 1500
     };
 
 }
diff --git a/src/options.cpp b/src/options.cpp
index cd07c44..804cb4f 100644
--- a/src/options.cpp
+++ b/src/options.cpp
@@ -24,6 +24,8 @@ zmq::options_t::options_t () :
     lwm (0),
     swap (0),
     mask (0),
-    affinity (0)
+    affinity (0),
+    rate (0),
+    recovery_ivl (0)
 {
 }
diff --git a/src/options.hpp b/src/options.hpp
index faf21b8..9f4a264 100644
--- a/src/options.hpp
+++ b/src/options.hpp
@@ -37,6 +37,12 @@ namespace zmq
         uint64_t mask;
         uint64_t affinity;
         std::string identity;
+
+        //  Maximum tranfer rate [kb/s].
+        uint32_t rate;
+
+        //  Reliability time interval [s].
+        uint32_t recovery_ivl;
     };
 
 }
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index a26c280..2384d80 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -17,6 +17,8 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
+#include <iostream>
+
 #include <string>
 #include <algorithm>
 
@@ -35,6 +37,7 @@
 #include "uuid.hpp"
 #include "pipe.hpp"
 #include "err.hpp"
+#include "platform.hpp"
 
 zmq::socket_base_t::socket_base_t (app_thread_t *parent_) :
     object_t (parent_),
@@ -145,6 +148,22 @@ int zmq::socket_base_t::setsockopt (int option_, const void *optval_,
         errno = ENOTSUP;
         return -1;
 
+    case ZMQ_RATE:
+        if (optvallen_ != sizeof (uint32_t)) {
+            errno = EINVAL;
+            return -1;
+        }
+        options.rate = *((int32_t*) optval_);
+        return 0;
+        
+    case ZMQ_RECOVERY_IVL:
+        if (optvallen_ != sizeof (uint32_t)) {
+            errno = EINVAL;
+            return -1;
+        }
+        options.recovery_ivl = *((int32_t*) optval_);
+        return 0;
+
     default:
         errno = EINVAL;
         return -1;
@@ -170,6 +189,21 @@ int zmq::socket_base_t::connect (const char *addr_)
     std::string session_name ("#");
     session_name += uuid_t ().to_string ();
 
+    //  Parse addr_ string.
+    std::string addr_type;
+    std::string addr_args;
+
+    std::string addr (addr_);
+    std::string::size_type pos = addr.find ("://");
+
+    if (pos == std::string::npos) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    addr_type = addr.substr (0, pos);
+    addr_args = addr.substr (pos + 3);
+
     //  Create the session.
     io_thread_t *io_thread = choose_io_thread (options.affinity);
     session_t *session = new session_t (io_thread, this, session_name.c_str (),
@@ -198,20 +232,36 @@ int zmq::socket_base_t::connect (const char *addr_)
     send_plug (session);
     send_own (this, session);
 
-    //  Create the connecter object. Supply it with the session name so that
-    //  it can bind the new connection to the session once it is established.
-    zmq_connecter_t *connecter = new zmq_connecter_t (
-        choose_io_thread (options.affinity), this, options,
-        session_name.c_str ());
-    int rc = connecter->set_address (addr_);
-    if (rc != 0) {
-        delete connecter;
-        return -1;
+    if (addr_type == "tcp") {
+
+        //  Create the connecter object. Supply it with the session name so that
+        //  it can bind the new connection to the session once it is established.
+        zmq_connecter_t *connecter = new zmq_connecter_t (
+            choose_io_thread (options.affinity), this, options,
+            session_name.c_str ());
+        int rc = connecter->set_address (addr_args.c_str ());
+        if (rc != 0) {
+            delete connecter;
+            return -1;
+        }
+        send_plug (connecter);
+        send_own (this, connecter);
+
+        return 0;
     }
-    send_plug (connecter);
-    send_own (this, connecter);
 
-    return 0;
+#if defined ZMQ_HAVE_OPENPGM
+    if (addr_type == "pgm") {
+        
+        zmq_assert (false);
+
+        return 0;
+    }
+#endif
+
+    //  Unknown address type.
+    errno = ENOTSUP;
+    return -1;
 }
 
 int zmq::socket_base_t::send (::zmq_msg_t *msg_, int flags_)