如何在C和Python兩端的nanomsg中設置Pub/Sub？

Question

我正試圖學習nanomsg庫。

我正在使用版本C和Python的代碼示例。我試圖用Python腳本來訂閱C服務，但沒有任何事情發生。

這裏是我的兩個代碼：

的Python用戶

from __future__ import print_function 
from nanomsg import Socket, PAIR, PUB 
s2 = Socket(PAIR) 
while(True): 
    s2.connect('tcp://127.0.0.1:5555') 
    s2.send(b'hello nanomsg #1') 
    s2.send(b'hello nanomsg #2') 
    s2.close() 

C代碼：

#include <stdio.h> 
#include <stdlib.h> 
#include <string.h> 
#include <time.h> 
#include <netinet/in.h> /* For htonl and ntohl */ 
#include <unistd.h> 

#include <nanomsg/nn.h> 
#include <nanomsg/pubsub.h> 

/* The server runs forever. */ 
int server(const char *url) 
{ 
    int fd; 

    /* Create the socket. */ 
    fd = nn_socket (AF_SP, NN_PUB); 
    if (fd < 0) { 
     fprintf (stderr, "nn_socket: %s\n", nn_strerror (nn_errno())); 
     return (-1); 
    } 

    /* Bind to the URL. This will bind to the address and listen 
     synchronously; new clients will be accepted asynchronously 
     without further action from the calling program. */ 

    if (nn_bind (fd, url) < 0) { 
     fprintf (stderr, "nn_bind: %s\n", nn_strerror (nn_errno())); 
     nn_close (fd); 
     return (-1); 
    } 

    /* Now we can just publish results. Note that there is no explicit 
     accept required. We just start writing the information. */ 

    for (;;) { 
     uint8_t msg[2 * sizeof (uint32_t)]; 
     uint32_t secs, subs; 
     int rc; 

     secs = (uint32_t) time (NULL); 
     subs = (uint32_t) nn_get_statistic (fd, NN_STAT_CURRENT_CONNECTIONS); 

     secs = htonl (secs); 
     subs = htonl (subs); 

     memcpy (msg, &secs, sizeof (secs)); 
     memcpy (msg + sizeof (secs), &subs, sizeof (subs)); 

     rc = nn_send (fd, msg, sizeof (msg), 0); 
     if (rc < 0) { 
      /* There are several legitimate reasons this can fail. 
       We note them for debugging purposes, but then ignore 
       otherwise. */ 
      fprintf (stderr, "nn_send: %s (ignoring)\n", 
       nn_strerror (nn_errno())); 
     } 
     sleep(10); 
    } 

    /* NOTREACHED */ 
    nn_close (fd); 
    return (-1); 
} 

/* The client runs in a loop, displaying the content. */ 
int client (const char *url) 
{ 
    int fd; 
    int rc; 

    fd = nn_socket (AF_SP, NN_SUB); 
    if (fd < 0) { 
     fprintf (stderr, "nn_socket: %s\n", nn_strerror (nn_errno())); 
     return (-1); 
    } 

    if (nn_connect (fd, url) < 0) { 
     fprintf (stderr, "nn_socket: %s\n", nn_strerror (nn_errno())); 
     nn_close (fd); 
     return (-1);   
    } 

    /* We want all messages, so just subscribe to the empty value. */ 
    if (nn_setsockopt (fd, NN_SUB, NN_SUB_SUBSCRIBE, "", 0) < 0) { 
     fprintf (stderr, "nn_setsockopt: %s\n", nn_strerror (nn_errno())); 
     nn_close (fd); 
     return (-1);   
    } 

    for (;;) { 
     uint8_t msg[2 * sizeof (uint32_t)]; 
     char hhmmss[9]; /* HH:MM:SS\0 */ 
     uint32_t subs, secs; 
     time_t t; 

     rc = nn_recv (fd, msg, sizeof (msg), 0); 
     if (rc < 0) { 
      fprintf (stderr, "nn_recv: %s\n", nn_strerror (nn_errno())); 
      break; 
     } 
     if (rc != sizeof (msg)) { 
      fprintf (stderr, "nn_recv: got %d bytes, wanted %d\n", 
       rc, (int)sizeof (msg)); 
      break; 
     } 
     memcpy (&secs, msg, sizeof (secs)); 
     memcpy (&subs, msg + sizeof (secs), sizeof (subs)); 

     t = (time_t) ntohl(secs); 
     strftime (hhmmss, sizeof (hhmmss), "%T", localtime (&t)); 

     printf ("%s <pid %u> There are %u clients connected.\n", hhmmss, 
      (unsigned) getpid(), (unsigned) ntohl(subs)); 
    } 

    nn_close (fd); 
    return (-1); 
} 

int main (int argc, char **argv) 
{ 
    int rc; 
    if ((argc == 3) && (strcmp (argv[2], "-s") == 0)) { 
     rc = server (argv[1]); 
    } else if (argc == 2) { 
     rc = client (argv[1]); 
    } else { 
     fprintf (stderr, "Usage: %s <url> [-s]\n", argv[0]); 
     exit (EXIT_FAILURE); 
    } 
    exit (rc == 0 ? EXIT_SUCCESS : EXIT_FAILURE); 
} 

我做

./pubsub_demo tcp://127.0.0.1:5555 -s

運行C代碼

感謝您的幫助

Answer 1

C代碼看起來不錯。它來自here。

簡單版本的C NN_PUB服務器和NN_SUB客戶端也是exists。

提出Python代碼有幾個問題。

1）在nanomsg我們必須匹配behavioural-「協議」。爲了從C服務器接收到NN_PUB廣播，我們必須在Python端有一個匹配的套接字，而不是PAIR套接字。

2）連接到相同的endpoint- transport-class://address:port作爲NN_PUB插座nn_bind() -s到。循環中沒有必要這樣做。

3）插座必須有SUB_SUBSCRIBE選項設置。

4）SUB套接字是用於收聽，它沒有被設計爲.send()任何東西。

一個未經考驗的Python程序看起來可能在原則如下：

# import appropriate modules for the nanomsg socket 
from nanomsg import Socket, PUB, SUB, SUB_SUBSCRIBE 

# open Python's SUB socket matching the NN_PUB socket on the C side 
s2 = Socket(SUB) 
# s2 should be >= 0 

# connect the socket to the same endpoint as NN_PUB server 
ret1 = s2.connect('tcp://127.0.0.1:5555') 
# ret1 should be 0 

# subscribe to everything: 
ret2 = s2.set_string_option(SUB, SUB_SUBSCRIBE, '') 
# ret1 should be 0 

# receive messages: 
while(True): 
    message = s2.recv() 

你也可以看看的Python測試PUB/SUB example

我希望它能幫助。