1 files changed, 38 insertions, 0 deletions
diff --git a/src/session_base.cpp b/src/session_base.cpp
index 12d2e8d..81f7347 100644
--- a/src/session_base.cpp
+++ b/src/session_base.cpp
@@ -30,6 +30,8 @@
 #include "ipc_connecter.hpp"
 #include "pgm_sender.hpp"
 #include "pgm_receiver.hpp"
+#include "udp_sender.hpp"
+#include "udp_receiver.hpp"
 
 #include "req.hpp"
 #include "xreq.hpp"
@@ -484,6 +486,42 @@ void xs::session_base_t::start_connecting (bool wait_)
     }
 #endif
 
+    //  UDP support.
+    if (protocol == "udp") {
+
+        //  At this point we'll create message pipes to the session straight
+        //  away. There's no point in delaying it as no concept of 'connect'
+        //  exists with UDP anyway.
+        if (options.type == XS_PUB || options.type == XS_XPUB) {
+
+            //  UDP sender.
+            udp_sender_t *udp_sender =  new (std::nothrow) udp_sender_t (
+                io_thread, options);
+            alloc_assert (udp_sender);
+
+            int rc = udp_sender->init (address.c_str ());
+            xs_assert (rc == 0);
+
+            send_attach (this, udp_sender);
+        }
+        else if (options.type == XS_SUB || options.type == XS_XSUB) {
+
+            //  UDP receiver.
+            udp_receiver_t *udp_receiver =  new (std::nothrow) udp_receiver_t (
+                io_thread, options);
+            alloc_assert (udp_receiver);
+
+            int rc = udp_receiver->init (address.c_str ());
+            xs_assert (rc == 0);
+
+            send_attach (this, udp_receiver);
+        }
+        else
+            xs_assert (false);
+
+        return;
+    }
+
     xs_assert (false);
 }