summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/pgm_receiver.cpp25
-rw-r--r--src/pgm_receiver.hpp2
-rw-r--r--src/zmq_decoder.cpp10
-rw-r--r--src/zmq_decoder.hpp2
4 files changed, 16 insertions, 23 deletions
diff --git a/src/pgm_receiver.cpp b/src/pgm_receiver.cpp
index f34ecf0..6ea310c 100644
--- a/src/pgm_receiver.cpp
+++ b/src/pgm_receiver.cpp
@@ -46,6 +46,7 @@
zmq::pgm_receiver_t::pgm_receiver_t (class io_thread_t *parent_,
const options_t &options_, const char *session_name_) :
io_object_t (parent_),
+ decoder (NULL),
pgm_socket (true, options_),
options (options_),
session_name (session_name_),
@@ -56,10 +57,15 @@ zmq::pgm_receiver_t::pgm_receiver_t (class io_thread_t *parent_,
zmq::pgm_receiver_t::~pgm_receiver_t ()
{
+ if (decoder)
+ delete decoder;
}
int zmq::pgm_receiver_t::init (const char *network_)
{
+ decoder = new zmq_decoder_t;
+ zmq_assert (decoder);
+
return pgm_socket.init (network_);
}
@@ -69,7 +75,7 @@ void zmq::pgm_receiver_t::plug (i_inout *inout_)
int socket_fd;
int waiting_pipe_fd;
- decoder.set_inout (inout_);
+ decoder->set_inout (inout_);
// Fill socket_fd and waiting_pipe_fd from PGM transport
pgm_socket.get_receiver_fds (&socket_fd, &waiting_pipe_fd);
@@ -91,7 +97,7 @@ void zmq::pgm_receiver_t::unplug ()
{
rm_fd (socket_handle);
rm_fd (pipe_handle);
- decoder.set_inout (NULL);
+ decoder->set_inout (NULL);
inout = NULL;
}
@@ -105,9 +111,14 @@ void zmq::pgm_receiver_t::reconnect ()
// Save inout ptr.
i_inout *inout_tmp = inout;
+ // PGM receiver is not joined anymore.
+ joined = false;
+
// Unplug - plug PGM transport.
unplug ();
- decoder.reset ();
+ delete decoder;
+ decoder = new zmq_decoder_t;
+ zmq_assert (decoder);
plug (inout_tmp);
}
@@ -121,7 +132,7 @@ void zmq::pgm_receiver_t::in_event ()
while ((nbytes = receive_with_offset (&data_with_offset)) > 0) {
// Push all the data to the decoder.
- decoder.write ((unsigned char*)data_with_offset, nbytes);
+ decoder->write ((unsigned char*)data_with_offset, nbytes);
}
// Flush any messages decoder may have produced to the dispatcher.
@@ -130,12 +141,6 @@ void zmq::pgm_receiver_t::in_event ()
// Data loss detected.
if (nbytes == -1) {
- // Throw message in progress from decoder
- decoder.reset ();
-
- // PGM receive is not joined anymore.
- joined = false;
-
// Recreate PGM transport.
reconnect ();
}
diff --git a/src/pgm_receiver.hpp b/src/pgm_receiver.hpp
index ce9fa1a..53d5340 100644
--- a/src/pgm_receiver.hpp
+++ b/src/pgm_receiver.hpp
@@ -64,7 +64,7 @@ namespace zmq
ssize_t receive_with_offset (void **data_);
// Message decoder.
- zmq_decoder_t decoder;
+ zmq_decoder_t *decoder;
// PGM socket.
pgm_socket_t pgm_socket;
diff --git a/src/zmq_decoder.cpp b/src/zmq_decoder.cpp
index ccaad0d..53811a1 100644
--- a/src/zmq_decoder.cpp
+++ b/src/zmq_decoder.cpp
@@ -36,16 +36,6 @@ zmq::zmq_decoder_t::~zmq_decoder_t ()
zmq_msg_close (&in_progress);
}
-void zmq::zmq_decoder_t::reset ()
-{
- // Free and reinit message buffer.
- zmq_msg_close (&in_progress);
- zmq_msg_init (&in_progress);
-
- // Restart the FSM.
- next_step (tmpbuf, 1, &zmq_decoder_t::one_byte_size_ready);
-}
-
void zmq::zmq_decoder_t::set_inout (i_inout *destination_)
{
destination = destination_;
diff --git a/src/zmq_decoder.hpp b/src/zmq_decoder.hpp
index 6bc373b..16b5312 100644
--- a/src/zmq_decoder.hpp
+++ b/src/zmq_decoder.hpp
@@ -37,8 +37,6 @@ namespace zmq
void set_inout (struct i_inout *destination_);
- // Clears any partially decoded messages.
- void reset ();
private:
bool one_byte_size_ready ();