[C]: reflecting changes for issue #377.

Author: tmontgomery

aeron-driver/src/main/c/aeron_ipc_publication.c

@@ -95,7 +95,7 @@ int aeron_ipc_publication_create(
     _pub->conductor_fields.subscribeable.array = NULL;
     _pub->conductor_fields.subscribeable.length = 0;
     _pub->conductor_fields.subscribeable.capacity = 0;
-    _pub->conductor_fields.subscribeable.add_position_hook_func = aeron_ipc_publication_add_subscriber_hook;
+    _pub->conductor_fields.subscribeable.add_position_hook_func = aeron_driver_subscribeable_null_hook;
     _pub->conductor_fields.subscribeable.remove_position_hook_func = aeron_ipc_publication_remove_subscriber_hook;
     _pub->conductor_fields.subscribeable.clientd = _pub;
     _pub->conductor_fields.managed_resource.registration_id = registration_id;
@@ -238,10 +238,12 @@ void aeron_ipc_publication_decref(void *clientd)
 
 void aeron_ipc_publication_check_for_blocked_publisher(aeron_ipc_publication_t *publication, int64_t now_ns)
 {
-    if (publication->conductor_fields.consumer_position == publication->conductor_fields.last_consumer_position)
+    int64_t consumer_position = publication->conductor_fields.consumer_position;
+
+    if (consumer_position == publication->conductor_fields.last_consumer_position &&
+        aeron_ipc_publication_producer_position(publication) > consumer_position)
     {
-        if (now_ns > (publication->conductor_fields.time_of_last_consumer_position_change + publication->unblock_timeout_ns) &&
-            aeron_ipc_publication_producer_position(publication) > publication->conductor_fields.consumer_position)
+        if (now_ns > (publication->conductor_fields.time_of_last_consumer_position_change + publication->unblock_timeout_ns))
         {
             if (aeron_logbuffer_unblocker_unblock(
                 publication->mapped_raw_log.term_buffers,
@@ -259,7 +261,6 @@ void aeron_ipc_publication_check_for_blocked_publisher(aeron_ipc_publication_t *
     }
 }
 
-extern void aeron_ipc_publication_add_subscriber_hook(void *clientd, int64_t *value_addr);
 extern void aeron_ipc_publication_remove_subscriber_hook(void *clientd, int64_t *value_addr);
 extern int64_t aeron_ipc_publication_producer_position(aeron_ipc_publication_t *publication);
 extern int64_t aeron_ipc_publication_joining_position(aeron_ipc_publication_t *publication);

aeron-driver/src/main/c/aeron_ipc_publication.h

@@ -99,17 +99,6 @@ void aeron_ipc_publication_decref(void *clientd);
 
 void aeron_ipc_publication_check_for_blocked_publisher(aeron_ipc_publication_t *publication, int64_t now_ns);
 
-inline void aeron_ipc_publication_add_subscriber_hook(void *clientd, int64_t *value_addr)
-{
-    aeron_ipc_publication_t *publication = (aeron_ipc_publication_t *)clientd;
-
-    /* 0 to 1 transition */
-    if (0 == publication->conductor_fields.subscribeable.length)
-    {
-        publication->conductor_fields.time_of_last_consumer_position_change = publication->nano_clock();
-    }
-}
-
 inline void aeron_ipc_publication_remove_subscriber_hook(void *clientd, int64_t *value_addr)
 {
     aeron_ipc_publication_t *publication = (aeron_ipc_publication_t *)clientd;

aeron-driver/src/main/c/aeron_network_publication.c

@@ -637,10 +637,9 @@ int aeron_network_publication_update_pub_lmt(aeron_network_publication_t *public
 void aeron_network_publication_check_for_blocked_publisher(
     aeron_network_publication_t *publication, int64_t now_ns, int64_t snd_pos)
 {
-    if (snd_pos == publication->conductor_fields.last_snd_pos)
+    if (snd_pos == publication->conductor_fields.last_snd_pos && aeron_network_publication_producer_position(publication) > snd_pos)
     {
-        if (now_ns > (publication->conductor_fields.time_of_last_activity_ns + publication->unblock_timeout_ns) &&
-            aeron_network_publication_producer_position(publication) > snd_pos)
+        if (now_ns > (publication->conductor_fields.time_of_last_activity_ns + publication->unblock_timeout_ns))
         {
             if (aeron_logbuffer_unblocker_unblock(
                 publication->mapped_raw_log.term_buffers,