使用ZeroMQ接收C中的多部分消息

我正在爲使用ZeroMQ套接字傳輸數據併發布多部分消息的設備寫入數據記錄系統。使用ZeroMQ接收C中的多部分消息

我的程序將需要C.

要寫入我曾經試圖寫一個Python腳本，其工作完全，因爲它應該。這看起來是這樣的：

import zmq 
import time 
import numpy as np 

_HEARTBEAT_PREFIX =  b'ray.heartbeat\0' 
_RAW_DATA_PREFIX =  b'ray.rawdata\0' 

context = zmq.Context.instance() 

time.sleep(5) 

socket = context.socket(zmq.SUB) 
socket.setsockopt(zmq.LINGER, 0) 
socket.setsockopt(zmq.SUBSCRIBE, b'') 
time.sleep(5) 
socket.connect("tcp://localhost:50290") 
time.sleep(5) 
go = True 
while go: 

    if socket.poll(1000): 

     zframes = socket.recv_multipart(flags=zmq.NOBLOCK, copy=False) 

     if (zframes[0].buffer.tobytes() == _HEARTBEAT_PREFIX): 
      print ('Heartbeart') 
     elif (zframes[0].buffer.tobytes() == _RAW_DATA_PREFIX): 
      dataToWrite = np.frombuffer(zframes[1].buffer, dtype='i2') 
      print ('Raw data: ') 
      # and so on ...

這是我第一次用ZMQ工作，並試圖端口這個Python代碼C是一個真正的挑戰。 C代碼我現在已經寫運行，但不能正常工作：

#include "zhelpers.h" 

int main (int argc, char *argv []) 
{ 

    void *context = zmq_ctx_new(); 
    void *subscriber = zmq_socket(context,ZMQ_SUB); 

    int rc = zmq_connect(subscriber,"tcp://localhost:50290"); 
    assert(rc == 0); 
    rc = zmq_setsockopt(subscriber,ZMQ_SUBSCRIBE, "", 0); 
    assert(rc == 0); 

    while (1) 
    { 
    zmq_msg_t message; 
    zmq_msg_init (&message); 
    zmq_msg_recv (&message, subscriber,0);//, ZMQ_NOBLOCK); 
    // Process the message frame 


    int size = zmq_msg_size(&message); 
    char *string = malloc(size + 1); 
    memcpy(string, zmq_msg_data(&message), size); 


    zmq_msg_close (&message); 

    string[size] = 0; 
    printf("Message[%d]: %s\n", size, string); 

    if (!zmq_msg_more (&message)) 
     break;  // Last message frame 
    } 
    return 0; 
}

此代碼將然而，提供輸入數據的大小正確，但顯示的數據時，它顯示爲亂碼（符號等on ..）我將需要像在Python中那樣將傳入的數據作爲字符串來進一步處理它。

來源

2016-07-22 Jesper

C代碼似乎沒有檢查幀類型（心跳/原始數據）。第一幀似乎是類型，如果類型是原始數據，則下一幀包含可打印的內容。 –

我發現Python中的numpy數組被轉換爲一個包含int的字符串數組，並且他們在C程序中解釋了非常困難的值。

通過操縱接收到的數據我能夠提取正確的價值觀：

#include "zhelpers.h" 

int main (int argc, char *argv []) 
{ 

    int i = 0; 

    void *context = zmq_ctx_new(); 
    void *subscriber = zmq_socket(context,ZMQ_SUB); 

    int rc = zmq_connect(subscriber,"tcp://localhost:50290"); 
    assert(rc == 0); 

    char* filter = "ray."; 

    rc = zmq_setsockopt(subscriber,ZMQ_SUBSCRIBE, filter, sizeof(filter)); 
    assert(rc == 0); 
    int nextFrame = 0; 

    while (1) 
    { 

    zmq_msg_t message; 
    zmq_msg_init (&message); 
    zmq_msg_recv (&message, subscriber,0);//, ZMQ_NOBLOCK); 

    // Process the message frame 

    int size = zmq_msg_size(&message); 
    char *string = malloc(size + 1); 
    memcpy(string, zmq_msg_data(&message), size); 

    if (nextFrame) 
    { 

     zmq_msg_close (&message); 
     printf("Raw data: "); 
     string[size] = 0; 
     //printf("Data: %s\n", string); 
     for (i = 0; i < size; i+=2) 
     { 

     printf("%d ", (int16_t) ((string[i] & 0xff) +((string[i+1] & 0xff)<<8))); 

     } 


    } 
     if (strcmp(string, "ray.rawdata") != 0) 
     { 
     nextFrame = 0; 
     break; 
     } 
     else 
     { 
     nextFrame = 1; 

     zmq_msg_close (&message); 

     string[size] = 0; 
     printf("Prefix: %s\n", string); 

     if (!zmq_msg_more (&message)) 
      break;  // Last message frame 
     } 

    }} 
    return 0; 
}

注意，這僅僅是接收從Python的numpy的陣列時，一個問題 - 只要它是一個心跳字符串或什麼的，沒有問題。

來源

2016-07-22 11:54:22 Jesper

使用ZeroMQ接收C中的多部分消息

回答

相關問題