libzmq  master
ZeroMQ C++ Core Engine (LIBZMQ)
test_proxy_terminate.cpp
Go to the documentation of this file.
1 /*
2  Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
3 
4  This file is part of libzmq, the ZeroMQ core engine in C++.
5 
6  libzmq is free software; you can redistribute it and/or modify it under
7  the terms of the GNU Lesser General Public License (LGPL) as published
8  by the Free Software Foundation; either version 3 of the License, or
9  (at your option) any later version.
10 
11  As a special exception, the Contributors give you permission to link
12  this library with independent modules to produce an executable,
13  regardless of the license terms of these independent modules, and to
14  copy and distribute the resulting executable under terms of your choice,
15  provided that you also meet, for each linked independent module, the
16  terms and conditions of the license of that module. An independent
17  module is a module which is not derived from or based on this library.
18  If you modify this library, you must extend this exception to your
19  version of the library.
20 
21  libzmq is distributed in the hope that it will be useful, but WITHOUT
22  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
23  FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
24  License for more details.
25 
26  You should have received a copy of the GNU Lesser General Public License
27  along with this program. If not, see <http://www.gnu.org/licenses/>.
28 */
29 
30 #include "testutil.hpp"
31 
32 // This is a test for issue #1382. The server thread creates a SUB-PUSH
33 // steerable proxy. The main process then sends messages to the SUB
34 // but there is no pull on the other side, previously the proxy blocks
35 // in writing to the backend, preventing the proxy from terminating
36 
37 void
38 server_task (void *ctx)
39 {
40  // Frontend socket talks to main process
41  void *frontend = zmq_socket (ctx, ZMQ_SUB);
42  assert (frontend);
43  int rc = zmq_setsockopt (frontend, ZMQ_SUBSCRIBE, "", 0);
44  assert (rc == 0);
45  rc = zmq_bind (frontend, "tcp://127.0.0.1:15564");
46  assert (rc == 0);
47 
48  // Nice socket which is never read
49  void *backend = zmq_socket (ctx, ZMQ_PUSH);
50  assert (backend);
51  rc = zmq_bind (backend, "tcp://127.0.0.1:15563");
52  assert (rc == 0);
53 
54  // Control socket receives terminate command from main over inproc
55  void *control = zmq_socket (ctx, ZMQ_SUB);
56  assert (control);
57  rc = zmq_setsockopt (control, ZMQ_SUBSCRIBE, "", 0);
58  assert (rc == 0);
59  rc = zmq_connect (control, "inproc://control");
60  assert (rc == 0);
61 
62  // Connect backend to frontend via a proxy
63  rc = zmq_proxy_steerable (frontend, backend, NULL, control);
64  assert (rc == 0);
65 
66  rc = zmq_close (frontend);
67  assert (rc == 0);
68  rc = zmq_close (backend);
69  assert (rc == 0);
70  rc = zmq_close (control);
71  assert (rc == 0);
72 }
73 
74 
75 // The main thread simply starts a basic steerable proxy server, publishes some messages, and then
76 // waits for the server to terminate.
77 
78 int main (void)
79 {
81 
82  void *ctx = zmq_ctx_new ();
83  assert (ctx);
84  // Control socket receives terminate command from main over inproc
85  void *control = zmq_socket (ctx, ZMQ_PUB);
86  assert (control);
87  int rc = zmq_bind (control, "inproc://control");
88  assert (rc == 0);
89 
90  void *thread = zmq_threadstart(&server_task, ctx);
91  msleep (500); // Run for 500 ms
92 
93  // Start a secondary publisher which writes data to the SUB-PUSH server socket
94  void *publisher = zmq_socket (ctx, ZMQ_PUB);
95  assert (publisher);
96  rc = zmq_connect (publisher, "tcp://127.0.0.1:15564");
97  assert (rc == 0);
98 
100  rc = zmq_send (publisher, "This is a test", 14, 0);
101  assert (rc == 14);
102 
103  msleep (50);
104  rc = zmq_send (publisher, "This is a test", 14, 0);
105  assert (rc == 14);
106 
107  msleep (50);
108  rc = zmq_send (publisher, "This is a test", 14, 0);
109  assert (rc == 14);
110  rc = zmq_send (control, "TERMINATE", 9, 0);
111  assert (rc == 9);
112 
113  rc = zmq_close (publisher);
114  assert (rc == 0);
115  rc = zmq_close (control);
116  assert (rc == 0);
117 
118  zmq_threadclose (thread);
119 
120  rc = zmq_ctx_term (ctx);
121  assert (rc == 0);
122  return 0;
123 }
void msleep(int milliseconds)
Definition: testutil.hpp:316
ZMQ_EXPORT int zmq_setsockopt(void *s, int option, const void *optval, size_t optvallen)
Definition: zmq.cpp:265
ZMQ_EXPORT void * zmq_ctx_new(void)
Definition: zmq.cpp:115
#define ZMQ_SUB
Definition: zmq.h:248
#define ZMQ_SUBSCRIBE
Definition: zmq.h:266
void setup_test_environment(void)
Definition: testutil.hpp:285
#define SETTLE_TIME
Definition: testutil.hpp:44
#define ZMQ_PUB
Definition: zmq.h:247
ZMQ_EXPORT void * zmq_socket(void *, int type)
Definition: zmq.cpp:244
#define ZMQ_PUSH
Definition: zmq.h:254
ZMQ_EXPORT void zmq_threadclose(void *thread)
Definition: zmq_utils.cpp:85
ZMQ_EXPORT int zmq_connect(void *s, const char *addr)
Definition: zmq.cpp:332
ZMQ_EXPORT int zmq_close(void *s)
Definition: zmq.cpp:255
ZMQ_EXPORT int zmq_send(void *s, const void *buf, size_t len, int flags)
Definition: zmq.cpp:387
ZMQ_EXPORT int zmq_bind(void *s, const char *addr)
Definition: zmq.cpp:321
ZMQ_EXPORT int zmq_ctx_term(void *context)
Definition: zmq.cpp:162
ZMQ_EXPORT void * zmq_threadstart(zmq_thread_fn *func, void *arg)
Definition: zmq_utils.cpp:78
int main(void)
ZMQ_EXPORT int zmq_proxy_steerable(void *frontend, void *backend, void *capture, void *control)
Definition: zmq.cpp:1335
void server_task(void *ctx)