0
0
mirror of https://github.com/zeromq/libzmq.git synced 2025-01-15 02:07:59 +08:00

Merged fix for #1382

This commit is contained in:
Rik van der Heijden 2015-04-24 23:01:20 +02:00 committed by Pieter Hintjens
parent 3ef80cec74
commit f5e1c607a5
7 changed files with 124 additions and 3 deletions

1
.gitignore vendored
View File

@ -74,6 +74,7 @@ test_linger
test_security_null
test_security_plain
test_proxy
test_proxy_terminate
test_abstract_ipc
test_filter_ipc
test_connect_delay_tipc

View File

@ -86,6 +86,7 @@ Philip Kovacs
Pieter Hintjens
Piotr Trojanek
Richard Newton
Rik van der Heijden
Robert G. Jakabosky
Sebastian Otaegui
Stefan Radomski

View File

@ -320,6 +320,7 @@ test_apps = \
test_inproc_connect \
test_issue_566 \
test_proxy \
test_proxy_terminate \
test_many_sockets \
test_ipc_wildcard \
test_diffserv \

4
NEWS
View File

@ -1,4 +1,4 @@
0MQ version 4.1.1 rc2, released on 2014/12/xx
0MQ version 4.1.1 rc2, released on 2015/xx/xx
=============================================
* Fixed #1208 - fix recursion in automake packaging
@ -17,6 +17,8 @@
* Fixed #1389 - PUB, PUSH sockets had slow memory leak.
* Fixed #1382 - zmq_proxy did not terminate if there were no readers.
0MQ version 4.1.0 rc1, released on 2014/10/14
=============================================

View File

@ -159,14 +159,16 @@ int zmq::proxy (
}
// Process a request
if (state == active
&& items [0].revents & ZMQ_POLLIN) {
&& items [0].revents & ZMQ_POLLIN
&& items [1].revents & ZMQ_POLLOUT) {
rc = forward(frontend_, backend_, capture_,msg);
if (unlikely (rc < 0))
return -1;
}
// Process a reply
if (state == active
&& items [1].revents & ZMQ_POLLIN) {
&& items [1].revents & ZMQ_POLLIN
&& items [0].revents & ZMQ_POLLOUT) {
rc = forward(backend_, frontend_, capture_,msg);
if (unlikely (rc < 0))
return -1;

View File

@ -52,6 +52,7 @@ if(NOT WIN32)
test_reqrep_ipc
test_abstract_ipc
test_proxy
test_proxy_terminate
test_filter_ipc
)
if(HAVE_FORK)

View File

@ -0,0 +1,113 @@
/*
Copyright (c) 2007-2015 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "testutil.hpp"
#include "../include/zmq_utils.h"
// This is a test for issue #1382. The server thread creates a SUB-PUSH
// steerable proxy. The main process then sends messages to the SUB
// but there is no pull on the other side, previously the proxy blocks
// in writing to the backend, preventing the proxy from terminating
void
server_task (void *ctx)
{
// Frontend socket talks to main process
void *frontend = zmq_socket (ctx, ZMQ_SUB);
assert (frontend);
int rc = zmq_setsockopt (frontend, ZMQ_SUBSCRIBE, "", 0);
assert (rc == 0);
rc = zmq_bind (frontend, "tcp://127.0.0.1:15564");
assert (rc == 0);
// Nice socket which is never read
void *backend = zmq_socket (ctx, ZMQ_PUSH);
assert (backend);
rc = zmq_bind (backend, "tcp://127.0.0.1:15563");
assert (rc == 0);
// Control socket receives terminate command from main over inproc
void *control = zmq_socket (ctx, ZMQ_SUB);
assert (control);
rc = zmq_setsockopt (control, ZMQ_SUBSCRIBE, "", 0);
assert (rc == 0);
rc = zmq_connect (control, "inproc://control");
assert (rc == 0);
// Connect backend to frontend via a proxy
zmq_proxy_steerable (frontend, backend, NULL, control);
rc = zmq_close (frontend);
assert (rc == 0);
rc = zmq_close (backend);
assert (rc == 0);
rc = zmq_close (control);
assert (rc == 0);
}
// The main thread simply starts a basic steerable proxy server, publishes some messages, and then
// waits for the server to terminate.
int main (void)
{
setup_test_environment ();
void *ctx = zmq_ctx_new ();
assert (ctx);
// Control socket receives terminate command from main over inproc
void *control = zmq_socket (ctx, ZMQ_PUB);
assert (control);
int rc = zmq_bind (control, "inproc://control");
assert (rc == 0);
void *thread = zmq_threadstart(&server_task, ctx);
msleep (500); // Run for 500 ms
// Start a secondary publisher which writes data to the SUB-PUSH server socket
void *publisher = zmq_socket (ctx, ZMQ_PUB);
assert (publisher);
rc = zmq_connect (publisher, "tcp://127.0.0.1:15564");
assert (rc == 0);
msleep (50);
rc = zmq_send (publisher, "This is a test", 14, 0);
assert (rc == 14);
msleep (50);
rc = zmq_send (publisher, "This is a test", 14, 0);
assert (rc == 14);
msleep (50);
rc = zmq_send (publisher, "This is a test", 14, 0);
assert (rc == 14);
rc = zmq_send (control, "TERMINATE", 9, 0);
assert (rc == 9);
rc = zmq_close (publisher);
assert (rc == 0);
rc = zmq_close (control);
assert (rc == 0);
zmq_threadclose (thread);
rc = zmq_ctx_term (ctx);
assert (rc == 0);
return 0;
}