ZMQ REQ Message Source: Difference between revisions

From GNU Radio
Jump to navigation Jump to search
(Created page with "Category:Block Docs Category:Stub Docs This is the template for the "Page-per-block Docs". This first section should describe what the block...")
 
(add note about IP addresses)
 
(8 intermediate revisions by 2 users not shown)
Line 1: Line 1:
[[Category:Block Docs]]
[[Category:Block Docs]]
[[Category:Stub Docs]]
The ZMQ REQ Message Source block receives messages from a ZMQ REQ socket and outputs async messages. This block will connect to a ZMQ REP Message Sink.<br>
This is the template for the [[:Category:Block_Docs|"Page-per-block Docs"]]. This first section should describe what the block does and how to use it, using however many paragraphs necessary.  Note that the title of the wiki page should match the block's name in GRC, i.e. the one defined in the block's .grc file.  Look at the [[FFT]] Block for a good example.
The zeromq.org website says:<br>
 
"The REQ-REP socket pair is in lockstep. The client issues zmq_send() and then zmq_recv(), in a loop (or once if that's all it needs). Doing any other sequence (e.g., sending two messages in a row) will result in a return code of -1 from the send or recv call." Likewise, the server "issues zmq_recv() and then zmq_send() in that order, as often as it needs to."<br>
As this is a basic template, it's also in the [[:Category:Stub_Docs|"Stub Docs category"]]. Please improve it.


== Parameters ==
== Parameters ==
(''R''): <span class="plainlinks">[https://wiki.gnuradio.org/index.php/GNURadioCompanion#Variable_Controls ''Run-time adjustable'']</span>
(''R''): <span class="plainlinks">[https://wiki.gnuradio.org/index.php/GNURadioCompanion#Variable_Controls ''Run-time adjustable'']</span>


; Param 1 (''R'')
; Address
: Description of parameter, provide any tips or recommended values. Note that the name of the parameter above should match the param's label that shows up in grc (e.g. Sample Rate).
: ZMQ socket address specifier. The format of the address is <code>tcp://*:port</code> where * should be 127.0.0.1 for localhost.<br>
: <b>Note:</b> If the Source and Sink blocks are on two different computers on the same LAN, then the IP and port number of the Sink block must be specified on each end of that connection. For example, if the Sink is on IP 192.168.2.14:5678 and the Source is on IP 192.168.2.5, both Source and Sink blocks must specify the Sink IP and port (192.168.2.14:5678).


; Param 2
; Timeout
: blah blah blah
: Socket timeout in milliseconds, default is 100ms.


== Example Flowgraph ==
== Example Flowgraph ==


Insert description of flowgraph here, then show a screenshot of the flowgraph and the output if there is an interesting GUI. Currently we have no standard method of uploading the actual flowgraph to the wiki or git repo, unfortunately. The plan is to have an example flowgraph showing how the block might be used, for every block, and the flowgraphs will live in the git repo.
=== Inter-flowgraph ===
 
Request/Reply pairs can be used on one, or two separate, flowgraphs to exchange messages.<br>
 
[[File:REQ_REP_msg_demo_fg.png]]
 
=== External Python client (send only) ===
 
An external Python program can send messages to a ZMQ REQ Message Source block. An example flowgraph and Python code follow.<br>
 
[[File:Msg_test7_fg.png]]
 
The Python client code looks like this:<br>
 
<pre>
#!/usr/bin/python3
# -*- coding: utf-8 -*-
 
# zmqREP.py
#
#  The REQest / REPly nomenclature of the GNU Radio message blocks is from
#  the perspective of the flowgraph. So, to send a 'request' to GNU Radio, the message
#  must be sent as a 'reply' from the Python client, Likewise, a 'reply' from GNU Radio
#  must be received as a 'request' to the Python client! Therefore, send on the reply socket
#  and receive on the request socket.
#
#  The zeromq.org website says:
#  "The REQ-REP socket pair is in lockstep. The client issues zmq_send() and then zmq_recv(),
#  in a loop (or once if that's all it needs). Doing any other sequence (e.g., sending two messages in a row)
#  will result in a return code of -1 from the send or recv call." Likewise, the server "issues zmq_recv()
#  and then zmq_send() in that order, as often as it needs to."
#
#  To conform to that requirement, a non-standard "kludge" is used (see below).
 
import datetime
import time
import signal
import pmt
import zmq
 
_debug = 1          # set to zero to turn off diagnostics
 
# create socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_PORT = ":50123"
_ADDR = _PROTOCOL + _SERVER + _PORT
if (_debug):
    print ("zmqREP connecting to:", _ADDR)
context = zmq.Context()
if (_debug):
    assert (context)
sock = context.socket (zmq.REP)
if (_debug):
    assert (sock)
rc = sock.bind (_ADDR)
if (_debug):
    assert (rc == None)
 
while True:
    _tim = datetime.datetime.now()
    _out = str (_tim)
    try:
        sock.recv()                                # this is the non-standard "kludge"
        sock.send (pmt.serialize_str(pmt.to_pmt(_out)))      # send on the 'reply' socket
        time.sleep(5)
    except KeyboardInterrupt:
        if (_debug):
            print (" Interrupt received. shutting down.")
        # clean up
        sock.close()
        context.term()
        exit()
</pre>
 
=== GNU Radio as a server ===
 
If the GNU Radio flowgraph(s) is configured as a server, the REQ message is processed by the flowgraph and a message is sent back in a REP message as the response. An example flowgraph and Python code follow. Note that two different ports are used.<br>
 
[[File:Server_demo_fg.png]]
 
The Embedded Python block "Server demo" contains the following code:<br>
 
<pre>
from gnuradio import gr
import pmt
 
class my_sync_block (gr.sync_block):
#  accepts message string from input port
#  capitalizes the string
#  sends message to output port
    def __init__(self):
        gr.sync_block.__init__(self,
            name = "Server demo",
            in_sig = None,
            out_sig = None)
        self.message_port_register_in(pmt.intern('msg_in'))
        self.message_port_register_out(pmt.intern('msg_out'))
        self.set_msg_handler(pmt.intern('msg_in'), self.handle_msg)
 
    def handle_msg(self, msg):
        inputText = pmt.symbol_to_string (msg)
        # print (inputText)
        if (len (inputText) > 0):
            #  capitalize the string
            outputText = inputText.upper()
            # print (outputText)
            #  Send reply back to client
            self.message_port_pub(pmt.intern('msg_out'), pmt.intern(outputText))
 
    def work(self, input_items, output_items):
        # with no data ports, there is nothing to do
        return (0)
</pre>
 
The Python client code looks like this:<br>
 
<pre>
#!/usr/bin/python3
# -*- coding: utf-8 -*-
 
# zmq_REQ_client.py
#
#  The REQest / REPly nomenclature of the GNU Radio message blocks is from
#  the perspective of the flowgraph. So, to send a 'request' to GNU Radio, the message
#  must be sent as a 'reply' from the Python client, Likewise, a 'reply' from GNU Radio
#  must be received as a 'request' to the Python client! Therefore, send on the reply socket
#  and receive on the request socket.
#
#  The zeromq.org website says:
#  "The REQ-REP socket pair is in lockstep. The client issues zmq_send() and then zmq_recv(),
#  in a loop (or once if that's all it needs). Doing any other sequence (e.g., sending two messages in a row)
#  will result in a return code of -1 from the send or recv call." Likewise, the server "issues zmq_recv()
#  and then zmq_send() in that order, as often as it needs to."
#
#  To conform to that requirement, a non-standard "kludge" is used (see below).
 
import datetime
import time
import signal
import pmt
import zmq
 
_debug = 1          # set to zero to turn off diagnostics
 
# create a REQ socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_REQ_PORT = ":49202"
_REQ_ADDR = _PROTOCOL + _SERVER + _REQ_PORT
if (_debug):
    print ("'zmq_REQ_client' connecting to:", _REQ_ADDR)
req_context = zmq.Context()
if (_debug):
    assert (req_context)
req_sock = req_context.socket (zmq.REQ)
if (_debug):
    assert (req_sock)
rc = req_sock.connect (_REQ_ADDR)
if (_debug):
    assert (rc == None)
 
# create a REP socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_REP_PORT = ":49201"
_REP_ADDR = _PROTOCOL + _SERVER + _REP_PORT
if (_debug):
    print ("'zmq_REQ_client' binding to:", _REP_ADDR)
rep_context = zmq.Context()
if (_debug):
    assert (rep_context)
rep_sock = rep_context.socket (zmq.REP)
if (_debug):
    assert (rep_sock)
rc = rep_sock.bind (_REP_ADDR)
if (_debug):
    assert (rc == None)
 
while True:
    # generate an outgoing message
    _tim = datetime.datetime.now()
    _out = ("Local time is " + str (_tim))
    try:
        print(_out)
        rep_sock.recv()                                        # this is the non-standard "kludge"
        rep_sock.send (pmt.serialize_str(pmt.to_pmt(_out)))    # send on the 'reply' socket
        time.sleep(1)
        req_sock.send_string("\x01\x00\x00\x00")        # this is the non-standard "kludge"
        msg = req_sock.recv()                          # receive on the 'request' socket
        print (pmt.to_python(pmt.deserialize_str(msg)))
        time.sleep(4)
    except KeyboardInterrupt:
        if (_debug):
            print (" Interrupt received. shutting down.")
        # clean up
        req_sock.close()
        req_context.term()
        rep_sock.close()
        rep_context.term()
        exit()
</pre>


== Source Files ==
== Source Files ==

Latest revision as of 15:02, 26 March 2020

The ZMQ REQ Message Source block receives messages from a ZMQ REQ socket and outputs async messages. This block will connect to a ZMQ REP Message Sink.
The zeromq.org website says:
"The REQ-REP socket pair is in lockstep. The client issues zmq_send() and then zmq_recv(), in a loop (or once if that's all it needs). Doing any other sequence (e.g., sending two messages in a row) will result in a return code of -1 from the send or recv call." Likewise, the server "issues zmq_recv() and then zmq_send() in that order, as often as it needs to."

Parameters

(R): Run-time adjustable

Address
ZMQ socket address specifier. The format of the address is tcp://*:port where * should be 127.0.0.1 for localhost.
Note: If the Source and Sink blocks are on two different computers on the same LAN, then the IP and port number of the Sink block must be specified on each end of that connection. For example, if the Sink is on IP 192.168.2.14:5678 and the Source is on IP 192.168.2.5, both Source and Sink blocks must specify the Sink IP and port (192.168.2.14:5678).
Timeout
Socket timeout in milliseconds, default is 100ms.

Example Flowgraph

Inter-flowgraph

Request/Reply pairs can be used on one, or two separate, flowgraphs to exchange messages.

REQ REP msg demo fg.png

External Python client (send only)

An external Python program can send messages to a ZMQ REQ Message Source block. An example flowgraph and Python code follow.

Msg test7 fg.png

The Python client code looks like this:

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# zmqREP.py
#
#   The REQest / REPly nomenclature of the GNU Radio message blocks is from
#   the perspective of the flowgraph. So, to send a 'request' to GNU Radio, the message
#   must be sent as a 'reply' from the Python client, Likewise, a 'reply' from GNU Radio
#   must be received as a 'request' to the Python client! Therefore, send on the reply socket
#   and receive on the request socket.
#
#   The zeromq.org website says:
#   "The REQ-REP socket pair is in lockstep. The client issues zmq_send() and then zmq_recv(),
#   in a loop (or once if that's all it needs). Doing any other sequence (e.g., sending two messages in a row)
#   will result in a return code of -1 from the send or recv call." Likewise, the server "issues zmq_recv() 
#   and then zmq_send() in that order, as often as it needs to."
#
#   To conform to that requirement, a non-standard "kludge" is used (see below).

import datetime
import time
import signal
import pmt
import zmq

_debug = 1          # set to zero to turn off diagnostics

# create socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_PORT = ":50123"
_ADDR = _PROTOCOL + _SERVER + _PORT
if (_debug):
    print ("zmqREP connecting to:", _ADDR)
context = zmq.Context()
if (_debug):
    assert (context)
sock = context.socket (zmq.REP)
if (_debug):
    assert (sock)
rc = sock.bind (_ADDR)
if (_debug):
    assert (rc == None)

while True:
    _tim = datetime.datetime.now()
    _out = str (_tim)
    try:
        sock.recv()                                # this is the non-standard "kludge"
        sock.send (pmt.serialize_str(pmt.to_pmt(_out)))       # send on the 'reply' socket
        time.sleep(5)
    except KeyboardInterrupt:
        if (_debug):
            print (" Interrupt received. shutting down.")
        # clean up
        sock.close()
        context.term()
        exit()

GNU Radio as a server

If the GNU Radio flowgraph(s) is configured as a server, the REQ message is processed by the flowgraph and a message is sent back in a REP message as the response. An example flowgraph and Python code follow. Note that two different ports are used.

Server demo fg.png

The Embedded Python block "Server demo" contains the following code:

from gnuradio import gr
import pmt

class my_sync_block (gr.sync_block):
#  accepts message string from input port
#  capitalizes the string
#  sends message to output port
    def __init__(self):
        gr.sync_block.__init__(self,
            name = "Server demo",
            in_sig = None,
            out_sig = None)
        self.message_port_register_in(pmt.intern('msg_in'))
        self.message_port_register_out(pmt.intern('msg_out'))
        self.set_msg_handler(pmt.intern('msg_in'), self.handle_msg)

    def handle_msg(self, msg):
        inputText = pmt.symbol_to_string (msg)
        # print (inputText)
        if (len (inputText) > 0):
            #  capitalize the string
            outputText = inputText.upper()
            # print (outputText)
            #  Send reply back to client
            self.message_port_pub(pmt.intern('msg_out'), pmt.intern(outputText))

    def work(self, input_items, output_items):
        # with no data ports, there is nothing to do
        return (0)

The Python client code looks like this:

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# zmq_REQ_client.py
#
#   The REQest / REPly nomenclature of the GNU Radio message blocks is from
#   the perspective of the flowgraph. So, to send a 'request' to GNU Radio, the message
#   must be sent as a 'reply' from the Python client, Likewise, a 'reply' from GNU Radio
#   must be received as a 'request' to the Python client! Therefore, send on the reply socket
#   and receive on the request socket.
#
#   The zeromq.org website says:
#   "The REQ-REP socket pair is in lockstep. The client issues zmq_send() and then zmq_recv(),
#   in a loop (or once if that's all it needs). Doing any other sequence (e.g., sending two messages in a row)
#   will result in a return code of -1 from the send or recv call." Likewise, the server "issues zmq_recv() 
#   and then zmq_send() in that order, as often as it needs to."
#
#   To conform to that requirement, a non-standard "kludge" is used (see below).

import datetime
import time
import signal
import pmt
import zmq

_debug = 1          # set to zero to turn off diagnostics

# create a REQ socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_REQ_PORT = ":49202"
_REQ_ADDR = _PROTOCOL + _SERVER + _REQ_PORT
if (_debug):
    print ("'zmq_REQ_client' connecting to:", _REQ_ADDR)
req_context = zmq.Context()
if (_debug):
    assert (req_context)
req_sock = req_context.socket (zmq.REQ)
if (_debug):
    assert (req_sock)
rc = req_sock.connect (_REQ_ADDR)
if (_debug):
    assert (rc == None)

# create a REP socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_REP_PORT = ":49201"
_REP_ADDR = _PROTOCOL + _SERVER + _REP_PORT
if (_debug):
    print ("'zmq_REQ_client' binding to:", _REP_ADDR)
rep_context = zmq.Context()
if (_debug):
    assert (rep_context)
rep_sock = rep_context.socket (zmq.REP)
if (_debug):
    assert (rep_sock)
rc = rep_sock.bind (_REP_ADDR)
if (_debug):
    assert (rc == None)

while True:
    # generate an outgoing message
    _tim = datetime.datetime.now()
    _out = ("Local time is " + str (_tim))
    try:
        print(_out)
        rep_sock.recv()                                        # this is the non-standard "kludge"
        rep_sock.send (pmt.serialize_str(pmt.to_pmt(_out)))    # send on the 'reply' socket
        time.sleep(1)
        req_sock.send_string("\x01\x00\x00\x00")        # this is the non-standard "kludge"
        msg = req_sock.recv()                           # receive on the 'request' socket
        print (pmt.to_python(pmt.deserialize_str(msg)))
        time.sleep(4)
    except KeyboardInterrupt:
        if (_debug):
            print (" Interrupt received. shutting down.")
        # clean up
        req_sock.close()
        req_context.term()
        rep_sock.close()
        rep_context.term()
        exit()

Source Files

C++ files
TODO
Header files
TODO
Public header files
TODO
Block definition
TODO