Merge commit '31cbcd39be6aef3ed43121da5b797d8ec9b0fd31' as 'libs/cppzmq'

libs/cppzmq/examples/CMakeLists.txt

@@ -0,0 +1,40 @@
+cmake_minimum_required(VERSION 3.0 FATAL_ERROR)
+
+project(cppzmq-examples CXX)
+
+# place binaries and libraries according to GNU standards
+
+include(GNUInstallDirs)
+set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/${CMAKE_INSTALL_LIBDIR})
+set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/${CMAKE_INSTALL_LIBDIR})
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/${CMAKE_INSTALL_BINDIR})
+
+find_package(Threads)
+find_package(cppzmq)
+
+add_executable(
+    pubsub_multithread_inproc
+    pubsub_multithread_inproc.cpp
+)
+target_link_libraries(
+    pubsub_multithread_inproc
+    PRIVATE cppzmq ${CMAKE_THREAD_LIBS_INIT}
+)
+
+add_executable(
+    hello_world
+    hello_world.cpp
+)
+target_link_libraries(
+    hello_world
+    PRIVATE cppzmq ${CMAKE_THREAD_LIBS_INIT}
+)
+
+add_executable(
+    multipart_messages
+    multipart_messages.cpp
+)
+target_link_libraries(
+    multipart_messages
+    PRIVATE cppzmq ${CMAKE_THREAD_LIBS_INIT}
+)

libs/cppzmq/examples/hello_world.cpp

@@ -0,0 +1,9 @@
+#include <zmq.hpp>
+
+int main()
+{
+    zmq::context_t ctx;
+    zmq::socket_t sock(ctx, zmq::socket_type::push);
+    sock.bind("inproc://test");
+    sock.send(zmq::str_buffer("Hello, world"), zmq::send_flags::dontwait);
+}

libs/cppzmq/examples/multipart_messages.cpp

@@ -0,0 +1,31 @@
+#include <iostream>
+#include <zmq_addon.hpp>
+
+int main()
+{
+    zmq::context_t ctx;
+    zmq::socket_t sock1(ctx, zmq::socket_type::push);
+    zmq::socket_t sock2(ctx, zmq::socket_type::pull);
+    sock1.bind("tcp://127.0.0.1:*");
+    const std::string last_endpoint =
+        sock1.get(zmq::sockopt::last_endpoint);
+    std::cout << "Connecting to "
+              << last_endpoint << std::endl;
+    sock2.connect(last_endpoint);
+
+    std::array<zmq::const_buffer, 2> send_msgs = {
+        zmq::str_buffer("foo"),
+        zmq::str_buffer("bar!")
+    };
+    if (!zmq::send_multipart(sock1, send_msgs))
+        return 1;
+
+    std::vector<zmq::message_t> recv_msgs;
+    const auto ret = zmq::recv_multipart(
+        sock2, std::back_inserter(recv_msgs));
+    if (!ret)
+        return 1;
+    std::cout << "Got " << *ret
+              << " messages" << std::endl;
+    return 0;
+}

libs/cppzmq/examples/pubsub_multithread_inproc.cpp

@@ -0,0 +1,102 @@
+#include <future>
+#include <iostream>
+#include <string>
+#include <thread>
+
+#include "zmq.hpp"
+#include "zmq_addon.hpp"
+
+void PublisherThread(zmq::context_t *ctx) {
+    //  Prepare publisher
+    zmq::socket_t publisher(*ctx, zmq::socket_type::pub);
+    publisher.bind("inproc://#1");
+
+    // Give the subscribers a chance to connect, so they don't lose any messages
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+
+    while (true) {
+        //  Write three messages, each with an envelope and content
+        publisher.send(zmq::str_buffer("A"), zmq::send_flags::sndmore);
+        publisher.send(zmq::str_buffer("Message in A envelope"));
+        publisher.send(zmq::str_buffer("B"), zmq::send_flags::sndmore);
+        publisher.send(zmq::str_buffer("Message in B envelope"));
+        publisher.send(zmq::str_buffer("C"), zmq::send_flags::sndmore);
+        publisher.send(zmq::str_buffer("Message in C envelope"));
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+}
+
+void SubscriberThread1(zmq::context_t *ctx) {
+    //  Prepare subscriber
+    zmq::socket_t subscriber(*ctx, zmq::socket_type::sub);
+    subscriber.connect("inproc://#1");
+
+    //  Thread2 opens "A" and "B" envelopes
+    subscriber.set(zmq::sockopt::subscribe, "A");
+    subscriber.set(zmq::sockopt::subscribe, "B");
+
+    while (1) {
+        // Receive all parts of the message
+        std::vector<zmq::message_t> recv_msgs;
+        zmq::recv_result_t result =
+          zmq::recv_multipart(subscriber, std::back_inserter(recv_msgs));
+        assert(result && "recv failed");
+        assert(*result == 2);
+
+        std::cout << "Thread2: [" << recv_msgs[0].to_string() << "] "
+                  << recv_msgs[1].to_string() << std::endl;
+    }
+}
+
+void SubscriberThread2(zmq::context_t *ctx) {
+    //  Prepare our context and subscriber
+    zmq::socket_t subscriber(*ctx, zmq::socket_type::sub);
+    subscriber.connect("inproc://#1");
+
+    //  Thread3 opens ALL envelopes
+    subscriber.set(zmq::sockopt::subscribe, "");
+
+    while (1) {
+        // Receive all parts of the message
+        std::vector<zmq::message_t> recv_msgs;
+        zmq::recv_result_t result =
+          zmq::recv_multipart(subscriber, std::back_inserter(recv_msgs));
+        assert(result && "recv failed");
+        assert(*result == 2);
+
+        std::cout << "Thread3: [" << recv_msgs[0].to_string() << "] "
+                  << recv_msgs[1].to_string() << std::endl;
+    }
+}
+
+int main() {
+    /*
+     * No I/O threads are involved in passing messages using the inproc transport.
+     * Therefore, if you are using a ØMQ context for in-process messaging only you
+     * can initialise the context with zero I/O threads.
+     *
+     * Source: http://api.zeromq.org/4-3:zmq-inproc
+     */
+    zmq::context_t ctx(0);
+
+    auto thread1 = std::async(std::launch::async, PublisherThread, &ctx);
+
+    // Give the publisher a chance to bind, since inproc requires it
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+    auto thread2 = std::async(std::launch::async, SubscriberThread1, &ctx);
+    auto thread3 = std::async(std::launch::async, SubscriberThread2, &ctx);
+    thread1.wait();
+    thread2.wait();
+    thread3.wait();
+
+    /*
+     * Output:
+     *   An infinite loop of a mix of:
+     *     Thread2: [A] Message in A envelope
+     *     Thread2: [B] Message in B envelope
+     *     Thread3: [A] Message in A envelope
+     *     Thread3: [B] Message in B envelope
+     *     Thread3: [C] Message in C envelope
+     */
+}