Commit 81b8362a authored by Ian Barber's avatar Ian Barber

Revert "Fix a number of whitespace issues in various parts of the code, add…

Revert "Fix a number of whitespace issues in various parts of the code, add validation to most calls on the test and take a first stab at implementing the reconnection pipe blocking."

This reverts commit 6f6466f0.
parent 48d39776
......@@ -47,7 +47,6 @@ namespace zmq
own,
attach,
bind,
detach,
activate_read,
activate_write,
hiccup,
......@@ -88,11 +87,6 @@ namespace zmq
struct {
zmq::pipe_t *pipe;
} bind;
// Sent from session to socket to disconnect a pipe
struct {
zmq::pipe_t *pipe;
} detach;
// Sent by pipe writer to inform dormant pipe reader that there
// are messages in the pipe.
......
......@@ -92,11 +92,6 @@ void zmq::object_t::process_command (command_t &cmd_)
process_seqnum ();
break;
case command_t::detach:
process_detach (cmd_.args.detach.pipe);
process_seqnum ();
break;
case command_t::hiccup:
process_hiccup (cmd_.args.hiccup.pipe);
break;
......@@ -216,15 +211,6 @@ void zmq::object_t::send_bind (own_t *destination_, pipe_t *pipe_,
send_command (cmd);
}
void zmq::object_t::send_detach (own_t *destination_, pipe_t *pipe_)
{
command_t cmd;
cmd.destination = destination_;
cmd.type = command_t::detach;
cmd.args.detach.pipe = pipe_;
send_command (cmd);
}
void zmq::object_t::send_activate_read (pipe_t *destination_)
{
command_t cmd;
......@@ -345,11 +331,6 @@ void zmq::object_t::process_bind (pipe_t *pipe_)
zmq_assert (false);
}
void zmq::object_t::process_detach (pipe_t *pipe_)
{
zmq_assert (false);
}
void zmq::object_t::process_activate_read ()
{
zmq_assert (false);
......
......@@ -78,7 +78,6 @@ namespace zmq
zmq::i_engine *engine_, bool inc_seqnum_ = true);
void send_bind (zmq::own_t *destination_, zmq::pipe_t *pipe_,
bool inc_seqnum_ = true);
void send_detach (own_t *destination_, pipe_t *pipe_);
void send_activate_read (zmq::pipe_t *destination_);
void send_activate_write (zmq::pipe_t *destination_,
uint64_t msgs_read_);
......@@ -100,7 +99,6 @@ namespace zmq
virtual void process_own (zmq::own_t *object_);
virtual void process_attach (zmq::i_engine *engine_);
virtual void process_bind (zmq::pipe_t *pipe_);
virtual void process_detach (zmq::pipe_t *pipe_);
virtual void process_activate_read ();
virtual void process_activate_write (uint64_t msgs_read_);
virtual void process_hiccup (void *pipe_);
......
......@@ -228,7 +228,7 @@ void zmq::session_base_t::clean_pipes ()
}
void zmq::session_base_t::terminated (pipe_t *pipe_)
{
{
// Drop the reference to the deallocated pipe.
zmq_assert (pipe == pipe_);
pipe = NULL;
......@@ -306,15 +306,9 @@ void zmq::session_base_t::process_attach (i_engine *engine_)
zmq_assert (!pipe);
pipe = pipes [0];
// Remember the remote end of the pipe if required
if (options.delay_attach_on_connect == 1)
outpipe = pipes [1];
// Ask socket to plug into the pipe.
send_bind (socket, pipes [1]);
}
else if (outpipe && (options.delay_attach_on_connect == 1))
send_bind (socket, outpipe);
// Plug in the engine.
zmq_assert (!engine);
......@@ -411,11 +405,6 @@ void zmq::session_base_t::detached ()
// the socket object to resend all the subscriptions.
if (pipe && (options.type == ZMQ_SUB || options.type == ZMQ_XSUB))
pipe->hiccup ();
// For delayed connect situations, hiccup the socket to have it
// pause usage of this pipe
if (outpipe && options.delay_attach_on_connect == 1)
send_detach(socket, outpipe);
}
void zmq::session_base_t::start_connecting (bool wait_)
......
......@@ -52,6 +52,9 @@ namespace zmq
// To be used once only, when creating the session.
void attach_pipe (zmq::pipe_t *pipe_);
// To be used once only, for delayed connection
void onconnect_attach_pipe (pipe_t *pipe_);
// Following functions are the interface exposed towards the engine.
virtual int read (msg_t *msg_);
......@@ -103,10 +106,7 @@ namespace zmq
// Pipe connecting the session to its socket.
zmq::pipe_t *pipe;
// Socket end of pipe, in case of reconnection
zmq::pipe_t *outpipe;
// This flag is true if the remainder of the message being processed
// is still in the in pipe.
bool incomplete_in;
......
......@@ -529,13 +529,13 @@ int zmq::socket_base_t::connect (const char *addr_)
session_base_t *session = session_base_t::create (io_thread, true, this,
options, paddr);
errno_assert (session);
// PGM does not support subscription forwarding; ask for all data to be
// sent to this pipe.
bool icanhasall = false;
if (protocol == "pgm" || protocol == "epgm")
icanhasall = true;
if (options.delay_attach_on_connect != 1 && icanhasall != true) {
// Create a bi-directional pipe.
object_t *parents [2] = {this, session};
......@@ -547,7 +547,7 @@ int zmq::socket_base_t::connect (const char *addr_)
// Attach local end of the pipe to the socket object.
attach_pipe (pipes [0], icanhasall);
// Attach remote end of the pipe to the session object later on.
session->attach_pipe (pipes [1]);
}
......@@ -876,17 +876,6 @@ void zmq::socket_base_t::process_destroy ()
destroyed = true;
}
void zmq::socket_base_t::process_detach (pipe_t *pipe_)
{
// If we are blocking connecting threads, drop this one
if (options.delay_attach_on_connect == 1) {
zmq_assert (pipe_);
pipes.erase (pipe_);
// Let derived sockets know we're ditching this pipe
xterminated (pipe_);
}
}
int zmq::socket_base_t::xsetsockopt (int option_, const void *optval_,
size_t optvallen_)
{
......
......@@ -185,9 +185,6 @@ namespace zmq
void process_bind (zmq::pipe_t *pipe_);
void process_term (int linger_);
// Allow blocking reconnecting pipes
void process_detach (pipe_t *pipe_);
// Socket's mailbox object.
mailbox_t mailbox;
......
......@@ -36,33 +36,25 @@ int main (int argc, char *argv [])
int seen = 0;
void *context = zmq_ctx_new();
assert (context);
void *to = zmq_socket(context, ZMQ_PULL);
assert (to);
val = 0;
rc = zmq_setsockopt(to, ZMQ_LINGER, &val, sizeof(val));
assert (rc == 0);
rc = zmq_bind(to, "tcp://*:5555");
assert (rc == 0);
zmq_setsockopt(to, ZMQ_LINGER, &val, sizeof(val));
zmq_bind(to, "tcp://*:5555");
// Create a socket pushing to two endpoints - only 1 message should arrive.
void *from = zmq_socket (context, ZMQ_PUSH);
assert(from);
val = 0;
zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof(val));
rc = zmq_connect (from, "tcp://localhost:5556");
zmq_setsockopt(from, ZMQ_LINGER, &val, sizeof(val));
rc = zmq_connect(from, "tcp://localhost:5556");
assert (rc == 0);
rc = zmq_connect (from, "tcp://localhost:5555");
rc = zmq_connect(from, "tcp://localhost:5555");
assert (rc == 0);
for (int i = 0; i < 10; ++i)
{
std::string message("message ");
message += ('0' + i);
rc = zmq_send (from, message.data(), message.size(), 0);
assert(rc >= 0);
zmq_send(from, message.data(), message.size(), 0);
}
sleep(1);
......@@ -70,7 +62,7 @@ int main (int argc, char *argv [])
for (int i = 0; i < 10; ++i)
{
memset(&buffer, 0, sizeof(buffer));
rc = zmq_recv (to, &buffer, sizeof(buffer), ZMQ_DONTWAIT);
rc = zmq_recv(to, &buffer, sizeof(buffer), ZMQ_DONTWAIT);
if( rc == -1)
break;
seen++;
......@@ -89,39 +81,27 @@ int main (int argc, char *argv [])
context = zmq_ctx_new();
std::cout << " Rerunning with DELAY_ATTACH_ON_CONNECT\n";
to = zmq_socket (context, ZMQ_PULL);
assert (to);
rc = zmq_bind (to, "tcp://*:5560");
assert(rc == 0);
to = zmq_socket(context, ZMQ_PULL);
zmq_bind(to, "tcp://*:5560");
val = 0;
rc = zmq_setsockopt (to, ZMQ_LINGER, &val, sizeof(val));
assert (rc == 0);
zmq_setsockopt(to, ZMQ_LINGER, &val, sizeof(val));
// Create a socket pushing to two endpoints - all messages should arrive.
from = zmq_socket (context, ZMQ_PUSH);
assert (from);
val = 0;
rc = zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof(val));
assert (rc == 0);
zmq_setsockopt(from, ZMQ_LINGER, &val, sizeof(val));
val = 1;
rc = zmq_setsockopt (from, ZMQ_DELAY_ATTACH_ON_CONNECT, &val, sizeof(val));
zmq_setsockopt(from, ZMQ_DELAY_ATTACH_ON_CONNECT, &val, sizeof(val));
rc = zmq_connect(from, "tcp://localhost:5561");
assert (rc == 0);
rc = zmq_connect (from, "tcp://localhost:5561");
assert (rc == 0);
rc = zmq_connect (from, "tcp://localhost:5560");
rc = zmq_connect(from, "tcp://localhost:5560");
assert (rc == 0);
for (int i = 0; i < 10; ++i)
{
std::string message("message ");
message += ('0' + i);
rc = zmq_send (from, message.data(), message.size(), 0);
assert (rc >= 0);
zmq_send(from, message.data(), message.size(), 0);
}
sleep(1);
......@@ -130,9 +110,13 @@ int main (int argc, char *argv [])
for (int i = 0; i < 10; ++i)
{
memset(&buffer, 0, sizeof(buffer));
rc = zmq_recv (to, &buffer, sizeof(buffer), ZMQ_DONTWAIT);
assert (rc != -1);
rc = zmq_recv(to, &buffer, sizeof(buffer), ZMQ_DONTWAIT);
if( rc == -1) {
break;
}
seen++;
}
assert (seen == 10);
rc = zmq_close (from);
assert (rc == 0);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment