Posix updated to recent modularisation...

...of the NTF callback module.

Author: vveljko

lib/sockets/pbpal_ntf_callback_poller_poll.c

@@ -8,6 +8,8 @@
 #include "pubnub_assert.h"
 #include "pubnub_log.h"
 
+#include <string.h>
+
 
 #if defined(_WIN32)
 /* Yes, we do know that it's not really that simple, there are subtle,
@@ -17,6 +19,14 @@
 #define poll(fdarray, nfds, timeout) WSAPoll(fdarray, nfds, timeout)
 #endif
 
+#if !defined(INVALID_SOCKET)
+#define INVALID_SOCKET -1
+#endif
+
+#if !defined(SOCKET_ERROR)
+#define SOCKET_ERROR -1
+#endif
+
 
 struct pbpal_poll_data* pbpal_ntf_callback_poller_init(void)
 {

openssl/posix.mk

@@ -26,7 +26,7 @@ CFLAGS =-g -D PUBNUB_LOG_LEVEL=PUBNUB_LOG_LEVEL_ERROR  -Wall -D PUBNUB_THREADSAF
 # -g enables debugging, remove to get a smaller executable
 # -fsanitize=address Use AddressSanitizer
 
-INCLUDES=-I ../core -I . -I ../posix/fntest -I ../core/fntest -I ../lib/base64
+INCLUDES=-I ../core -I . -I ../posix/fntest -I ../core/fntest -I ../lib/base64 -I ../lib/sockets
 
 all: pubnub_sync_sample cancel_subscribe_sync_sample pubnub_sync_subloop_sample pubnub_callback_sample subscribe_publish_callback_sample pubnub_callback_subloop_sample pubnub_fntest pubnub_console_sync pubnub_console_callback pubnub_crypto_sync_sample
 
@@ -35,9 +35,12 @@ pubnub_sync.a : $(SOURCEFILES) ../core/pubnub_ntf_sync.c ../core/pubnub_sync_sub
 	$(CC) -c $(CFLAGS) $(INCLUDES) $(SOURCEFILES) ../core/pubnub_ntf_sync.c ../core/pubnub_sync_subscribe_loop.c
 	ar rcs pubnub_sync.a $(OBJFILES) pubnub_ntf_sync.o pubnub_sync_subscribe_loop.o
 
-pubnub_callback.a : $(SOURCEFILES)  ../core/pubnub_timer_list.c pubnub_ntf_callback_posix.c pubnub_get_native_socket.c ../core/pubnub_callback_subscribe_loop.c
-	$(CC) -c $(CFLAGS) $(INCLUDES) -D PUBNUB_CALLBACK_API $(SOURCEFILES)  ../core/pubnub_timer_list.c pubnub_ntf_callback_posix.c pubnub_get_native_socket.c ../core/pubnub_callback_subscribe_loop.c
-	ar rcs pubnub_callback.a $(OBJFILES)  pubnub_timer_list.o pubnub_ntf_callback_posix.o pubnub_get_native_socket.o pubnub_callback_subscribe_loop.o
+CALLBACK_INTF_SOURCEFILES=pubnub_ntf_callback_posix.c pubnub_get_native_socket.c ../core/pubnub_timer_list.c ../lib/sockets/pbpal_ntf_callback_poller_poll.c  ../core/pbpal_ntf_callback_queue.c ../core/pbpal_ntf_callback_admin.c ../core/pbpal_ntf_callback_handle_timer_list.c  ../core/pubnub_callback_subscribe_loop.c
+CALLBACK_INTF_OBJFILES=pubnub_ntf_callback_posix.o pubnub_get_native_socket.o pubnub_timer_list.o pbpal_ntf_callback_poller_poll.o pbpal_ntf_callback_queue.o pbpal_ntf_callback_admin.o pbpal_ntf_callback_handle_timer_list.o pubnub_callback_subscribe_loop.o
+
+pubnub_callback.a : $(SOURCEFILES) $(CALLBACK_INTF_SOURCEFILES)
+	$(CC) -c $(CFLAGS) $(INCLUDES) -D PUBNUB_CALLBACK_API $(SOURCEFILES) $(CALLBACK_INTF_SOURCEFILES)
+	ar rcs pubnub_callback.a $(OBJFILES) $(CALLBACK_INTF_OBJFILES)
 
 
 pubnub_sync_sample: ../core/samples/pubnub_sync_sample.c pubnub_sync.a

openssl/pubnub_ntf_callback_windows.c

@@ -32,7 +32,7 @@
 
 
 struct SocketWatcherData {
-    _Guarded_by_(mutw) struct pbpal_poll_data poll;
+    _Guarded_by_(mutw) struct pbpal_poll_data* poll;
     CRITICAL_SECTION mutw;
     HANDLE           thread_handle;
     DWORD            thread_id;
@@ -60,13 +60,13 @@ static int elapsed_ms(FILETIME prev_timspec, FILETIME timspec)
 
 int pbntf_watch_in_events(pubnub_t* pbp)
 {
-    return pbpal_ntf_watch_in_events(&m_watcher.poll, pbp);
+    return pbpal_ntf_watch_in_events(m_watcher.poll, pbp);
 }
 
 
 int pbntf_watch_out_events(pubnub_t* pbp)
 {
-    return pbpal_ntf_watch_out_events(&m_watcher.poll, pbp);
+    return pbpal_ntf_watch_out_events(m_watcher.poll, pbp);
 }
 
 
@@ -84,25 +84,9 @@ void socket_watcher_thread(void* arg)
 
         Sleep(1);
 
+        pbpal_ntf_poll_away(m_watcher.poll, ms);
+        
         EnterCriticalSection(&m_watcher.mutw);
-        if (m_watcher.poll.size > 0) {
-            int rslt = WSAPoll(m_watcher.poll.apoll, m_watcher.poll.size, ms);
-            if (SOCKET_ERROR == rslt) {
-                /* error? what to do about it? */
-                PUBNUB_LOG_WARNING("poll size = %ud, error = %d\n",
-                                   (unsigned)m_watcher.poll.size,
-                                   WSAGetLastError());
-            }
-            else if (rslt > 0) {
-                size_t i;
-                size_t apoll_size = m_watcher.poll.size;
-                for (i = 0; i < apoll_size; ++i) {
-                    if (m_watcher.poll.apoll[i].revents & (POLLIN | POLLOUT)) {
-                        pbntf_requeue_for_processing(m_watcher.poll.apb[i]);
-                    }
-                }
-            }
-        }
         if (PUBNUB_TIMERS_API) {
             FILETIME current_time;
             int      elapsed;
@@ -123,6 +107,10 @@ int pbntf_init(void)
 {
     InitializeCriticalSection(&m_watcher.mutw);
 
+    m_watcher.poll = pbpal_ntf_callback_poller_init();
+    if (NULL == m_watcher.poll) {
+        return -1;
+    }
     pbpal_ntf_callback_queue_init(&m_watcher.queue);
 
     m_watcher.thread_handle = (HANDLE)_beginthread(
@@ -132,6 +120,7 @@ int pbntf_init(void)
                          errno);
         DeleteCriticalSection(&m_watcher.mutw);
         pbpal_ntf_callback_queue_deinit(&m_watcher.queue);
+        pbpal_ntf_callback_poller_deinit(&m_watcher.poll);
         return -1;
     }
     m_watcher.thread_id = GetThreadId(m_watcher.thread_handle);
@@ -158,7 +147,7 @@ int pbntf_got_socket(pubnub_t* pb, pb_socket_t sockt)
 
     PUBNUB_UNUSED(sockt);
 
-    pbpal_ntf_callback_save_socket(&m_watcher.poll, pb);
+    pbpal_ntf_callback_save_socket(m_watcher.poll, pb);
     if (PUBNUB_TIMERS_API) {
         m_watcher.timer_head = pubnub_timer_list_add(m_watcher.timer_head, pb);
     }
@@ -177,7 +166,7 @@ void pbntf_lost_socket(pubnub_t* pb, pb_socket_t sockt)
 
     PUBNUB_UNUSED(sockt);
 
-    pbpal_ntf_callback_remove_socket(&m_watcher.poll, pb);
+    pbpal_ntf_callback_remove_socket(m_watcher.poll, pb);
     pbpal_remove_timer_safe(pb, &m_watcher.timer_head);
     pbpal_ntf_callback_remove_from_queue(&m_watcher.queue, pb);
 
@@ -191,7 +180,7 @@ void pbntf_update_socket(pubnub_t* pb, pb_socket_t socket)
 
     EnterCriticalSection(&m_watcher.mutw);
 
-    pbpal_ntf_callback_update_socket(&m_watcher.poll, pb);
+    pbpal_ntf_callback_update_socket(m_watcher.poll, pb);
 
     LeaveCriticalSection(&m_watcher.mutw);
 }

posix/posix.mk

@@ -26,17 +26,20 @@ CFLAGS =-g -Wall -D PUBNUB_THREADSAFE -D PUBNUB_LOG_LEVEL=PUBNUB_LOG_LEVEL_NONE
 # -g enables debugging, remove to get a smaller executable
 # -fsanitize-address Use AddressSanitizer
 
-INCLUDES=-I ../core -I . -I fntest -I ../core/fntest -I ../lib/base64 -I ../lib/md5
+INCLUDES=-I ../core -I . -I fntest -I ../core/fntest -I ../lib/base64 -I ../lib/md5 -I ../lib/sockets
 
 all: pubnub_sync_sample cancel_subscribe_sync_sample pubnub_sync_subloop_sample pubnub_sync_publish_retry pubnub_callback_sample pubnub_callback_subloop_sample subscribe_publish_callback_sample pubnub_fntest pubnub_console_sync pubnub_console_callback
 
 pubnub_sync.a : $(SOURCEFILES) ../core/pubnub_ntf_sync.c ../core/pubnub_sync_subscribe_loop.c
 	$(CC) -c $(CFLAGS) $(INCLUDES) $(SOURCEFILES) ../core/pubnub_ntf_sync.c ../core/pubnub_sync_subscribe_loop.c
 	ar rcs pubnub_sync.a $(OBJFILES) pubnub_ntf_sync.o pubnub_sync_subscribe_loop.o
 
-pubnub_callback.a : $(SOURCEFILES)  ../core/pubnub_timer_list.c pubnub_ntf_callback_posix.c pubnub_get_native_socket.c ../lib/sockets/pbpal_adns_sockets.c ../core/pubnub_callback_subscribe_loop.c
-	$(CC) -c $(CFLAGS) -D PUBNUB_CALLBACK_API $(INCLUDES) $(SOURCEFILES)  ../core/pubnub_timer_list.c pubnub_ntf_callback_posix.c pubnub_get_native_socket.c ../lib/sockets/pbpal_adns_sockets.c ../core/pubnub_callback_subscribe_loop.c
-	ar rcs pubnub_callback.a $(OBJFILES)  pubnub_timer_list.o pubnub_ntf_callback_posix.o pubnub_get_native_socket.o pbpal_adns_sockets.o pubnub_callback_subscribe_loop.o
+CALLBACK_INTF_SOURCEFILES=pubnub_ntf_callback_posix.c pubnub_get_native_socket.c ../core/pubnub_timer_list.c ../lib/sockets/pbpal_ntf_callback_poller_poll.c  ../core/pbpal_ntf_callback_queue.c ../core/pbpal_ntf_callback_admin.c ../core/pbpal_ntf_callback_handle_timer_list.c  ../core/pubnub_callback_subscribe_loop.c
+CALLBACK_INTF_OBJFILES=pubnub_ntf_callback_posix.o pubnub_get_native_socket.o pubnub_timer_list.o pbpal_ntf_callback_poller_poll.o pbpal_ntf_callback_queue.o pbpal_ntf_callback_admin.o pbpal_ntf_callback_handle_timer_list.o pubnub_callback_subscribe_loop.o
+
+pubnub_callback.a : $(SOURCEFILES) $(CALLBACK_INTF_SOURCEFILES)
+	$(CC) -c $(CFLAGS) -D PUBNUB_CALLBACK_API $(INCLUDES) $(SOURCEFILES) $(CALLBACK_INTF_SOURCEFILES)
+	ar rcs pubnub_callback.a $(OBJFILES) $(CALLBACK_INTF_OBJFILES)
 
 pubnub_sync_sample: ../core/samples/pubnub_sync_sample.c pubnub_sync.a
 	$(CC) -o $@ $(CFLAGS) $(INCLUDES) ../core/samples/pubnub_sync_sample.c pubnub_sync.a $(LDLIBS)

posix/pubnub_get_native_socket.c

@@ -1,8 +1,10 @@
 /* -*- c-file-style:"stroustrup"; indent-tabs-mode: nil -*- */
 #include "pubnub_internal.h"
 
+#include "pubnub_get_native_socket.h"
 
-int pubnub_get_native_socket(pubnub_t *pb)
+
+pbpal_native_socket_t pubnub_get_native_socket(pubnub_t *pb)
 {
     return pb->pal.socket;
 }

posix/pubnub_get_native_socket.h

@@ -4,11 +4,11 @@
 
 #include "pubnub_internal.h"
 
+typedef int pbpal_native_socket_t;
 
 /** Returns the native socket under POSIX of the given Pubnub context.
-    On POSIX, native (BSD) socket is an integer handle.
  */
-int pubnub_get_native_socket(pubnub_t *pb);
+pbpal_native_socket_t pubnub_get_native_socket(pubnub_t *pb);
 
 
 #endif /* !defined INC_PUBNUB_GET_NATIVE_SOCKET*/

posix/pubnub_internal.h

@@ -6,6 +6,7 @@
 #include <unistd.h>
 #include <sys/socket.h>
 #include <netdb.h>
+#include <arpa/inet.h>
 #include <errno.h>