====== ZeroMQ ======
  * [[http://www.zeromq.org/|Página web del proyecto]]
  * [[http://zguide.zeromq.org/page:all|Guía del framework]]
  * [[http://zeromq.github.io/pyzmq/api/index.html|Documentación pyzmq]]
===== Uso de la librería =====

Implementar una capa de mensajes con ZeroMQ:
  - Escoger tipo de transporte:
    * TCP, basado en red: ''<nowiki>tcp://hostname:port</nowiki>''
    * INPROC, intraproceso (inter-thread/greenlet): ''<nowiki>inproc://name</nowiki>''
    * IPC, interproceso: ''<nowiki>ipc:///tmp/filename</nowiki>''
    * MULTICAST, multicast via PGM (con OpenPGM y sólo para PUB-SUB): ''<nowiki>pgm://interface:address:port</nowiki>'' y ''<nowiki>epgm://interface:address:port</nowiki>''
  - Indicar la infraestructura:
    * Queue, para el request/response messaging pattern
    * Forwarder, para el publish/subscribe messaging pattern.
    * Streamer, para un pipelined paralellized messaging pattern.
  - Escoger el ptarón de mensaje:
    * REQUEST/REPLY, bidireccional, basado en estados y con balanceador de carga.
    * PUBLISH/SUBSCRIBE, publica a varios receptores de golpe.
    * UPSTREAM/DOWNSTREAM, distribución de los datos en una pipeline.
    * PAIR, comunicación exclusiva entre pares.

Pasos para desarrollar:
  - Obtener el contexto de ZeroMQ.
  - A partir del contexto definir los sockets necesarios a partir de la arquitectura que usarán.
  - Si realizan la acción de escucha (elemento estable) llamaremos al método ''bind'' del socket. Si empezarán enviando (elemento dinámico) al ''connect''.

El contexto es el contenedor de los sockets para un proceso. Para inicializar el contexto se usa ''zmq_ctx_new()'' y para eliminarlo ''zmq_ctx_destroy()''.

Propiedades de la librería:
  * Podemos conectar un gran número de clientes a un servidor.
  * Los clientes guardarán los paquetes hasta que el servidor se inicie.
  * Podemos conectar un cliente a tantos servidores como queramos, los mensajes se repartirán entre ellos.

Notas:
  * Cuando indicamos ''*'' como dirección nos estamos refiriendo a que tome todas las interfaces. Por ejemplo: ''<nowiki>http://*:5555</nowiki>''
  * Si un servidor deja de funcionar y el cliente está aún conectado, el funcionamiento de este último no será estable.
  * ZeroMQ no se encarga de la codificación de mensajes. Podremos enviarlo en el formato que queramos, por ejemplo simple JSON o de forma binaria con BSON, Protocol Buffers o Thrift.
===== Conceptos, patrones y arquitecturas =====

==== Request-Reply ====
{{:fw:zeromq:fig2.png?direct&100|}} \\ 
Esta arquitectura consiste en que el cliente envía un mensaje con ''zmq_send()'' y luego recibe con ''zmq_recv()'', el servidor lo hace en el orden contrario. Estas acciones se lleban a cabo en loop y sin poder cambiarse de orden. \\ \\ 
Servidor: 
<code python>
import zmq
import time
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")
while True:
    message = socket.recv()
    # Do some 'work'
    socket.send("World")
</code>
Cliente:
<code python>
import zmq
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect ("tcp://localhost:5555")
for request in range (10):
    socket.send ("Hello")
    #  Get the reply.
    message = socket.recv()
    print "Received reply ", request, "[", message, "]"
</code>
==== Publish-Subscribers (Pub\Sub) ====
{{:fw:zeromq:fig4.png?direct&300|}} \\ 
Es unidireccional. En este patrón un servidor publica sin importar quien recibe. Enviará a todos los clientes y serán estos quienes filtren. \\ 
Por ejemplo el siguiente código envía eventos de los distintos participantes...
<code python>
import zmq
from random import choice
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:5000")
countries = ['netherlands','brazil','germany','portugal']
events = ['yellow card', 'red card', 'goal', 'corner', 'foul']
while True:
    msg = choice( countries ) +" "+ choice( events )
    print "->",msg
    socket.send( msg )
</code>
... Y el siguiente recibe de Netherlands y Germany.
<code python>
import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:5000")
socket.setsockopt(zmq.SUBSCRIBE, "netherlands")
socket.setsockopt(zmq.SUBSCRIBE, "germany")
while True:
    print  socket.recv()
</code>
Notas:
  * Un subscriber puede conectarse a tantos publishers como sea necesario.
  * Si un publisher no tiene subscribers los mensajes quedarán desechados.
  * TCP con esta estructura es lento.
  * Para realizar una subscripción a todos los mensajes haremos:
<code python>
socket.setsockopt(zmq.SUBSCRIBE, "")
</code>
==== Pipeline ====
Funciona como el Rep-Req sólo que en este no hay una respuesta desde el otro nodo. Es perfecto para los mensajes en paralelo. Imaginemos la siguiente estructura: \\ 
{{:fw:zeromq:fig5.png?direct&200|}} \\ 
En esta...
  * Un proceso ventilador (productor) produce tareas que pueden ser realizadas en paralelo.
  * Las tareas la reciben los workers.
  * Existe un sink que recoge los resultados de los workers.

Ventilador:
<code python>
import zmq
import random
import time
context = zmq.Context()

sender = context.socket(zmq.PUSH)
sender.bind("tcp://*:5557")

sink = context.socket(zmq.PUSH)
sink.connect("tcp://localhost:5558")

# The first message is "0" and signals start of batch
sink.send('0')

# Send 100 tasks
total_msec = 0
for task_nbr in range(100):
    workload = random.randint(1, 100)
    total_msec += workload
    sender.send(str(workload))
</code>
Worker:
<code python>
import sys
import time
import zmq

context = zmq.Context()

receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")

sender = context.socket(zmq.PUSH)
sender.connect("tcp://localhost:5558")

while True:
    s = receiver.recv()
    sys.stdout.write('.')
    time.sleep(int(s)*0.001)
    sender.send('')
</code>
Sink:
<code python>
import sys
import time
import zmq

context = zmq.Context()

receiver = context.socket(zmq.PULL)
receiver.bind("tcp://*:5558")

# Wait for start of batch
s = receiver.recv()

tstart = time.time()
# Process 100 confirmations
total_msec = 0
for task_nbr in range(100):
    s = receiver.recv()
    if task_nbr % 10 == 0:
        sys.stdout.write(':')
    else:
        sys.stdout.write('.')
tend = time.time()
print "Total elapsed time: %d msec" % ((tend-tstart)*1000)
</code>
==== Paired sockets ====
Estos son una comunicación uno a uno:
<code python>
import zmq
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.bind("tcp://127.0.0.1:5555")
</code>
<code python>
import zmq
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.connect("tcp://127.0.0.1:5555")
</code>
==== Devices ====
Podemos configurar un nodo como device. Esto permitirá mejorar el rendimiento del sistema. 
=== Queue ===
Se usa para mediar entre un REQ/REP: \\ 
{{:fw:zeromq:queue.png?direct&200|}} \\ 
Para poder llevarlo a cabo se utilizan dos nuevos tipos de sockets ''zmq.ROUTER'' y ''zmq.DEALER'' en el nodo queue. El socket ROUTER recibe un mensaje de una conexión REQ y añade un ID al inicio de dicho paquete, luego es enviado por el DEALER. El DEALER escoge una conexión REP y lo envía, cuando este es procesado es devuelto otra al ROUTER y enviado al socket REQ que inició la conexión. Todo esto es posible (y rompe el patrón REQ\REP) mediante el ID que el ROUTER inserta. \\ 
REQ:
<code python>
import sys
import zmq
context = zmq.Context()
for x in xrange(10):
    sock = context.socket(zmq.REQ)
    sock.connect(sys.argv[1])
    print 'REQ is', x,
    sock.send(str(x))
    print 'REP is', sock.recv()
</code>
REP:
<code python>
import sys
import zmq
context = zmq.Context()
while True:
    sock = context.socket(zmq.REP)
    sock.connect(sys.argv[1])
    x = sock.recv()
    print 'REQ is', x,
    reply = 'x-%s' % x
    sock.send(reply)
    print 'REP is', reply
</code>
QUEUE:
<code python>
import sys
import zmq
context = zmq.Context()
s1 = context.socket(zmq.ROUTER)
s2 = context.socket(zmq.DEALER)
s1.bind(sys.argv[1])
s2.bind(sys.argv[2])
zmq.device(zmq.QUEUE, s1, s2)
</code>


=== Forwarding ===
Media entre un grupo de nodos push\sub. Como este conjunto no se bloquea no es necesario usar un tipo de socket especial:
<code python>
import sys
import zmq
context = zmq.Context()
s1 = context.socket(zmq.SUB)
s2 = context.socket(zmq.PUB)
s1.bind(sys.argv[1])
s2.bind(sys.argv[2])
s1.setsockopt(zmq.SUBSCRIBE, '')
zmq.device(zmq.FORWARDER, s1, s2)
</code>
Ahora podemos conectarle tantos publishers y subscribers como queramos.

=== Streaming ===
En este caso es útil para los nodos PUSH\PULL.
<code python>
context = zmq.Context()
s1 = context.socket(zmq.PULL)
s2 = context.socket(zmq.PUSH)
s1.bind("tcp://*:5550")
s2.bind("tcp://*:5551")
zmq.device(zmq.STREAMER, s1, s2)
</code>
<code python>
import zmq
import random
import time
context = zmq.Context()
 
sender = context.socket(zmq.PUSH)
sender.connect("tcp://localhost:5550")

total_msec = 0
for task_nbr in range(100):
    workload = random.randint(1, 100)
    sender.send(str(workload))
</code>

<code python>
import sys
import time
import zmq
 
context = zmq.Context()
 
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5551")
 
while True:
    s = receiver.recv()
    sys.stdout.write('.')
    sys.stdout.flush()
    time.sleep(int(s)*0.001)
</code>
=== Crear tu propio Streamer & Forwarder ===
Como los dos son unidireccionales, lo único que tendremos que hacer es copiar los datos de un socket al otro:
<code python>
import sys
import zmq

def zmq_streaming_device(a, b):
    while True:
        msg = a.recv()
        more = a.getsockopt(zmq.RCVMORE)
        if more:
            b.send(msg, zmq.SNDMORE)
        else:
            b.send(msg)
        sys.stdout.write('.')
        sys.stdout.flush()

context = zmq.Context()
s1 = context.socket(zmq.PULL)
s2 = context.socket(zmq.PUSH)
s1.bind("tcp://*:5550")
s2.bind("tcp://*:5551")
zmq_streaming_device(s1, s2)
</code>
===== Patrones Request/Reply =====
===== Patrones Publish/Subscribe =====
===== Avanzado =====
==== Control de flujo ====
Cuando enviamos un mensaje y no hay receptor, lo que hace ZeroMQ es guardarlo en memoria hasta que el receptor esté activo. \\ 
Para evitar problemas de memoria ZeroMQ provee de la opción de socket high water mark (''zmq.HWM''). Si la añadimos a ''setsockopt'' podremos indicar cuantos mensajes quedarán guardados en memoria. \\ 
Con la opción ''zmq.SWAP'' podemos indicar un tamaño de disco donde guardar más mensajes una vez el HWM sea superado.
<code python>
sock = context.socket(zmq.PUSH)
sock.setsockopt(zmq.HWM, 1000)
sock.setsockopt(zmq.SWAP, 200*2**10)
sock.connect(sys.argv[1])
while True:
    sock.send(sys.argv[1] + ':' + time.ctime())
</code>
===== Notas =====
  * Saber la versión con ''zmq.zmq_version()''.
  * Para realizar conexiones no bloqueantes los usaremos con [[fw:gevent&#zeromq|greenlets]].
  * Para instalar la librería en el sistema: ''apt-get install libzmq-dev''.

==== Instalar en Ubuntu ====
Instalación de la librería del sistema:
<code>
$ sudo apt-get install python-dev libzmq-dev
</code>
Instalación en Python:
<code>
$ sudo pip install pyzmq
</code>
Recuerda que para utilizar ''pip'' has de tener instalado el paquete ''python-pip''
==== Instalar en Windows ====
Necesitarás agregar el paquete ''VS2010 C++ redistributable''.
==== Como... ====
=== Enviar\recibir varios paquetes ===
<code python>
sock.send(part1, zmq.SNDMORE)
sock.send(part2, zmq.SNDMORE)
sock.send(part3, zmq.SNDMORE)
sock.send(final)
</code>
<code python>
more = True
parts = []
while more:
    parts.append(sock.recv())
    more = sock.getsockopt(zmq.RCVMORE)
</code>
=== Sistema Productor\Consumidor ===
Productor:
<code python>
import zmq
import random
import time

context = zmq.Context()

sender = context.socket(zmq.PUSH)
sender.bind("tcp://*:5557")
sender.bind("tcp://*:5558")

print "Sending tasks to workers"
random.seed()
total_msec = 0
for task_nbr in range(100):
    workload = random.randint(1, 100)
    total_msec += workload
    sender.send(str(workload))

print "Total expected cost: %s msec" % total_msec
time.sleep(1)
</code>
Consumidor:
<code python>
import sys
import time
import zmq

context = zmq.Context()

receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")

while True:
    s = receiver.recv()
    sys.stdout.write('.'+s)
    sys.stdout.flush()
    # Do the work
    time.sleep(int(s)*0.001)
</code>

=== Enviar no bloqueante ===
  * Pero cuidado, el objeto sender no se elminará hasta que los datos hayan sido enviados.
<code python>
sender.send(jdata, zmq.NOBLOCK)
</code>