Allow packet_stream_reconnect() to cut offHEADmaster

 a connecting stream. Cleanup timer.

Signed-off-by: Andrew Opalach <andrew@akon.city>

1 files changed, 11 insertions, 8 deletions

diff --git a/src/packet_stream.c b/src/packet_stream.c
index 7868242..0a45b22 100644
--- a/src/packet_stream.c
+++ b/src/packet_stream.c
@@ -104,7 +104,7 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
     }
 }
 
-static bool stream_connected(struct nn_packet_stream *stream)
+static bool set_stream_connected(struct nn_packet_stream *stream)
 {
     stream->connect = PACKET_STREAM_CONNECTED;
     al_assert(stream->corked);
@@ -134,7 +134,7 @@ static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents)
             // Try again on next POLLOUT, not sure if this can actually happen.
             return;
         }
-        if (stream_connected(stream)) {
+        if (set_stream_connected(stream)) {
             ev_io_start(stream->loop->ev, &stream->revent);
         }
     }
@@ -201,7 +201,7 @@ void nn_packet_stream_from_socket(struct nn_packet_stream *stream, struct nn_eve
     s32 fd = nn_socket_get_fd(&stream->sock);
     ev_io_init_n(&stream->revent, stream_read_callback, fd, EV_READ);
     ev_io_init_n(&stream->wevent, stream_write_callback, fd, EV_WRITE);
-    if (stream_connected(stream)) {
+    if (set_stream_connected(stream)) {
         ev_io_start(stream->loop->ev, &stream->revent);
     }
 }
@@ -234,15 +234,20 @@ void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_l
     do_connect_internal(stream, addr, port);
 }
 
+// Keep in mind this stream will have had it's state reset in stop_internal().
 void nn_packet_stream_reconnect(struct nn_packet_stream *stream, str *addr, u16 port)
 {
+    if (stream->connect == PACKET_STREAM_CONNECTING) {
+        stop_write_internal(stream);
+        stop_internal(stream);
+    }
     al_assert(stream->connect == PACKET_STREAM_DISCONNECTED);
     do_connect_internal(stream, addr, port);
 }
 
-bool nn_packet_stream_connected(struct nn_packet_stream *stream)
+bool nn_packet_stream_set_connected(struct nn_packet_stream *stream)
 {
-    return stream_connected(stream);
+    return set_stream_connected(stream);
 }
 
 void nn_packet_stream_cork(struct nn_packet_stream *stream, bool cork)
@@ -314,10 +319,8 @@ void nn_packet_stream_return_packets(struct nn_packet_stream *stream, struct nn_
 
 void nn_packet_stream_disconnect(struct nn_packet_stream *stream)
 {
+    al_assert(stream->connect != PACKET_STREAM_DISCONNECTED);
     al_assert(stream->connect != PACKET_STREAM_DISCONNECTING);
-    // If reusing this stream, keep in mind it will have been set back
-    // to a default state in stop_internal().
-    if (stream->connect == PACKET_STREAM_DISCONNECTED) return;
     if (stream->direct) {
         struct nn_packet_stream *direct = stream->direct;
         direct->connect = PACKET_STREAM_DISCONNECTED;