Playing with RabbitMQ (part 2). Now with Python

Do you remember the las post about RabbitMQ? In that post we created a small wrapper library to use RabbitMQ with node and PHP. I also work with Python and I also want to use the same RabbitMQ wrapper here. With Python there’re several libraries to use Rabbit. I’ll use pika.

The idea is the same than the another post. I want to use queues, exchanges and RPCs. So let’s start with queues:

We can create a queue receiver called ‘queue.backend’

from rabbit import builder

server = {
    'host': 'localhost',
    'port': 5672,
    'user': 'guest',
    'pass': 'guest',
}

def onData(data):
    print data['aaa']

builder.queue('queue.backend', server).receive(onData)

and emit messages to the queue

from rabbit import builder

server = {
    'host': 'localhost',
    'port': 5672,
    'user': 'guest',
    'pass': 'guest',
}

queue = builder.queue('queue.backend', server)

queue.emit({"aaa": 1})
queue.emit({"aaa": 2})
queue.emit({"aaa": 3})

The library (as the PHP and ones). Uses a builder class to create our instances

from queue import Queue
from rpc import RPC
from exchange import Exchange

defaults = {
    'queue': {
        'queue': {
            'passive': False,
            'durable': True,
            'exclusive': False,
            'autoDelete': False,
            'nowait': False
        },
        'consumer': {
            'noLocal': False,
            'noAck': False,
            'exclusive': False,
            'nowait': False
        }
    },
    'exchange': {
        'exchange': {
            'passive': False,
            'durable': True,
            'autoDelete': True,
            'internal': False,
            'nowait': False
        },
        'queue': {
            'passive': False,
            'durable': True,
            'exclusive': False,
            'autoDelete': True,
            'nowait': False
        },
        'consumer': {
            'noLocal': False,
            'noAck': False,
            'exclusive': False,
            'nowait': False
        }
    },
    'rpc': {
        'queue': {
            'passive': False,
            'durable': True,
            'exclusive': False,
            'autoDelete': True,
            'nowait': False
        },
        'consumer': {
            'noLocal': False,
            'noAck': False,
            'exclusive': False,
            'nowait': False
        }
    }
}

def queue(name, server):
    conf = defaults['queue']
    conf['server'] = server

    return Queue(name, conf)

def rpc(name, server):
    conf = defaults['rpc']
    conf['server'] = server

    return RPC(name, conf)

def exchange(name, server):
    conf = defaults['exchange']
    conf['server'] = server

    return Exchange(name, conf)

And our Queue class

import pika
import json
import time

class Queue:
    def __init__(self, name, conf):
        self.name = name
        self.conf = conf

    def emit(self, data=None):
        credentials = pika.PlainCredentials(self.conf['server']['user'], self.conf['server']['pass'])
        connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.conf['server']['host'], port=self.conf['server']['port'], credentials=credentials))
        channel = connection.channel()

        queueConf = self.conf['queue']
        channel.queue_declare(queue=self.name, passive=queueConf['passive'], durable=queueConf['durable'], exclusive=queueConf['exclusive'], auto_delete=queueConf['autoDelete'])

        channel.basic_publish(exchange='', routing_key=self.name, body=json.dumps(data), properties=pika.BasicProperties(delivery_mode=2))
        connection.close()

    def receive(self, callback):
        credentials = pika.PlainCredentials(self.conf['server']['user'], self.conf['server']['pass'])
        connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.conf['server']['host'], port=self.conf['server']['port'], credentials=credentials))
        channel = connection.channel()

        queueConf = self.conf['queue']
        channel.queue_declare(queue=self.name, passive=queueConf['passive'], durable=queueConf['durable'], exclusive=queueConf['exclusive'], auto_delete=queueConf['autoDelete'])

        def _callback(ch, method, properties, body):
            callback(json.loads(body))
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print "%s %s::%s" % (time.strftime("[%d/%m/%Y-%H:%M:%S]", time.localtime(time.time())), self.name, body)

        print "%s Queue '%s' initialized" % (time.strftime("[%d/%m/%Y-%H:%M:%S]", time.localtime(time.time())), self.name)
        consumerConf = self.conf['consumer']
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(_callback, self.name, no_ack=consumerConf['noAck'], exclusive=consumerConf['exclusive'])

        channel.start_consuming()

We also want to use exchanges to emit messages without waiting for answers, just as a event broadcast. We can emit messages:

from rabbit import builder

server = {
    'host': 'localhost',
    'port': 5672,
    'user': 'guest',
    'pass': 'guest',
}

exchange = builder.exchange('process.log', server)

exchange.emit("xxx.log", "aaaa")
exchange.emit("xxx.log", ["11", "aaaa"])
exchange.emit("yyy.log", "aaaa")

And listen to messages

from rabbit import builder

server = {
    'host': 'localhost',
    'port': 5672,
    'user': 'guest',
    'pass': 'guest',
}

def onData(routingKey, data):
    print routingKey, data

builder.exchange('process.log', server).receive("yyy.log", onData)

That’s the class

import pika
import json
import time

class Exchange:
    def __init__(self, name, conf):
        self.name = name
        self.conf = conf

    def emit(self, routingKey, data=None):
        credentials = pika.PlainCredentials(self.conf['server']['user'], self.conf['server']['pass'])
        connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.conf['server']['host'], port=self.conf['server']['port'], credentials=credentials))
        channel = connection.channel()

        exchangeConf = self.conf['exchange']
        channel.exchange_declare(exchange=self.name, type='topic', passive=exchangeConf['passive'], durable=exchangeConf['durable'], auto_delete=exchangeConf['autoDelete'], internal=exchangeConf['internal'])
        channel.basic_publish(exchange=self.name, routing_key=routingKey, body=json.dumps(data))
        connection.close()

    def receive(self, bindingKey, callback):
        credentials = pika.PlainCredentials(self.conf['server']['user'], self.conf['server']['pass'])
        connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.conf['server']['host'], port=self.conf['server']['port'], credentials=credentials))
        channel = connection.channel()

        exchangeConf = self.conf['exchange']
        channel.exchange_declare(exchange=self.name, type='topic', passive=exchangeConf['passive'], durable=exchangeConf['durable'], auto_delete=exchangeConf['autoDelete'], internal=exchangeConf['internal'])

        queueConf = self.conf['queue']
        result = channel.queue_declare(passive=queueConf['passive'], durable=queueConf['durable'], exclusive=queueConf['exclusive'], auto_delete=queueConf['autoDelete'])
        queue_name = result.method.queue

        channel.queue_bind(exchange=self.name, queue=queue_name, routing_key=bindingKey)

        print "%s Exchange '%s' initialized" % (time.strftime("[%d/%m/%Y-%H:%M:%S]", time.localtime(time.time())), self.name)

        def _callback(ch, method, properties, body):
            callback(method.routing_key, json.loads(body))
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print "%s %s:::%s" % (time.strftime("[%d/%m/%Y-%H:%M:%S]", time.localtime(time.time())), self.name, body)

        consumerConf = self.conf['consumer']
        channel.basic_consume(_callback, queue=queue_name, no_ack=consumerConf['noAck'], exclusive=consumerConf['exclusive'])
        channel.start_consuming()

And finally we can use RPCs. Emit

from rabbit import builder

server = {
    'host': 'localhost',
    'port': 5672,
    'user': 'guest',
    'pass': 'guest',
}

print builder.rpc('rpc.hello', server).call("Gonzalo", "Ayuso")

And the server side

from rabbit import builder

server = {
    'host': 'localhost',
    'port': 5672,
    'user': 'guest',
    'pass': 'guest',
}

def onData(name, surname):
    return "Hello %s %s" % (name, surname)

builder.rpc('rpc.hello', server).server(onData)

And that’s the class

import pika
import json
import time
import uuid

class RPC:
    def __init__(self, name, conf):
        self.name = name
        self.conf = conf

    def call(self, *params):
        pika.PlainCredentials(self.conf['server']['user'], self.conf['server']['pass'])
        connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.conf['server']['host'], port=self.conf['server']['port']))
        channel = connection.channel()

        queueConf = self.conf['queue']
        result = channel.queue_declare(queue='', passive=queueConf['passive'], durable=queueConf['durable'], exclusive=queueConf['exclusive'], auto_delete=queueConf['autoDelete'])
        callback_queue = result.method.queue
        consumerConf = self.conf['consumer']
        channel.basic_consume(self.on_call_response, no_ack=consumerConf['noAck'], exclusive=consumerConf['exclusive'], queue='')

        self.response = None
        self.corr_id = str(uuid.uuid4())
        channel.basic_publish(exchange='', routing_key=self.name, properties=pika.BasicProperties(reply_to=callback_queue, correlation_id=self.corr_id), body=json.dumps(params))
        while self.response is None:
            connection.process_data_events()
        return self.response

    def on_call_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def server(self, callback):
        pika.PlainCredentials(self.conf['server']['user'], self.conf['server']['pass'])
        connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.conf['server']['host'], port=self.conf['server']['port']))
        channel = connection.channel()

        queueConf = self.conf['queue']
        channel.queue_declare(self.name, passive=queueConf['passive'], durable=queueConf['durable'], exclusive=queueConf['exclusive'], auto_delete=queueConf['autoDelete'])

        channel.basic_qos(prefetch_count=1)
        consumerConf = self.conf['consumer']

        def on_server_request(ch, method, props, body):
            response = callback(*json.loads(body))

            ch.basic_publish(exchange='', routing_key=props.reply_to, properties=pika.BasicProperties(correlation_id=props.correlation_id), body=json.dumps(response))
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print "%s %s::req => '%s' response => '%s'" % (time.strftime("[%d/%m/%Y-%H:%M:%S]", time.localtime(time.time())), self.name, body, response)

        channel.basic_consume(on_server_request, queue=self.name, no_ack=consumerConf['noAck'], exclusive=consumerConf['exclusive'])

        print "%s RPC '%s' initialized" % (time.strftime("[%d/%m/%Y-%H:%M:%S]", time.localtime(time.time())), self.name)
        channel.start_consuming()

And that’s all. Full project is available within my github account

Advertisements

Playing with RabbitMQ, PHP and node

I need to use RabbitMQ in one project. I’m a big fan of Gearman, but I must admit Rabbit is much more powerful. In this project I need to handle with PHP code and node, so I want to build a wrapper for those two languages. I don’t want to re-invent the wheel so I will use existing libraries (php-amqplib and amqplib for node).

Basically I need to use three things: First I need to create exchange channels to log different actions. I need to decouple those actions from the main code. I also need to create work queues to ensure those works are executed. It doesn’t matter if work is executed later but it must be executed. And finally RPC commands.

Let’s start with the queues. I want to push events to a queue in PHP

use G\Rabbit\Builder;
$server = [
    'host' => 'localhost',
    'port' => 5672,
    'user' => 'guest',
    'pass' => 'guest',
];
$queue = Builder::queue('queue.backend', $server);
$queue->emit(["aaa" => 1]);

and also with node

var server = {
    host: 'localhost',
    port: 5672,
    user: 'guest',
    pass: 'guest'
};

var queue = builder.queue('queue.backend', server);
queue.emit({aaa: 1});

And I also want to register workers to those queues with PHP and node

use G\Rabbit\Builder;
$server = [
    'host' => 'localhost',
    'port' => 5672,
    'user' => 'guest',
    'pass' => 'guest',
];
Builder::queue('queue.backend', $server)->receive(function ($data) {
    error_log(json_encode($data));
});
var server = {
    host: 'localhost',
    port: 5672,
    user: 'guest',
    pass: 'guest'
};

var queue = builder.queue('queue.backend', server);
queue.receive(function (data) {
    console.log(data);
});

Both implementations use one builder. In this case we are using Queue:

namespace G\Rabbit;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class Queue
{
    private $name;
    private $conf;
    public function __construct($name, $conf)
    {
        $this->name = $name;
        $this->conf = $conf;
    }
    private function createConnection()
    {
        $server = $this->conf['server'];
        return new AMQPStreamConnection($server['host'], $server['port'], $server['user'], $server['pass']);
    }
    private function declareQueue($channel)
    {
        $conf = $this->conf['queue'];
        $channel->queue_declare($this->name, $conf['passive'], $conf['durable'], $conf['exclusive'],
            $conf['auto_delete'], $conf['nowait']);
    }
    public function emit($data = null)
    {
        $connection = $this->createConnection();
        $channel = $connection->channel();
        $this->declareQueue($channel);
        $msg = new AMQPMessage(json_encode($data),
            ['delivery_mode' => 2] # make message persistent
        );
        $channel->basic_publish($msg, '', $this->name);
        $channel->close();
        $connection->close();
    }
    public function receive(callable $callback)
    {
        $connection = $this->createConnection();
        $channel = $connection->channel();
        $this->declareQueue($channel);
        $consumer = $this->conf['consumer'];
        if ($consumer['no_ack'] === false) {
            $channel->basic_qos(null, 1, null);
        }
        $channel->basic_consume($this->name, '', $consumer['no_local'], $consumer['no_ack'], $consumer['exclusive'],
            $consumer['nowait'],
            function ($msg) use ($callback) {
                call_user_func($callback, json_decode($msg->body, true), $this->name);
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
                $now = new \DateTime();
                echo '['.$now->format('d/m/Y H:i:s')."] {$this->name}::".$msg->body, "\n";
            });
        $now = new \DateTime();
        echo '['.$now->format('d/m/Y H:i:s')."] Queue '{$this->name}' initialized \n";
        while (count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }
}
var amqp = require('amqplib/callback_api');

var Queue = function (name, conf) {
    return {
        emit: function (data, close=true) {
            amqp.connect(`amqp://${conf.server.user}:${conf.server.pass}@${conf.server.host}:${conf.server.port}`, function (err, conn) {
                conn.createChannel(function (err, ch) {
                    var msg = JSON.stringify(data);

                    ch.assertQueue(name, conf.queue);
                    ch.sendToQueue(name, new Buffer(msg));
                });
                if (close) {
                    setTimeout(function () {
                        conn.close();
                        process.exit(0)
                    }, 500);
                }
            });
        },
        receive: function (callback) {
            amqp.connect(`amqp://${conf.server.user}:${conf.server.pass}@${conf.server.host}:${conf.server.port}`, function (err, conn) {
                conn.createChannel(function (err, ch) {
                    ch.assertQueue(name, conf.queue);
                    console.log(new Date().toString() + ' Queue ' + name + ' initialized');
                    ch.consume(name, function (msg) {
                        console.log(new Date().toString() + " Received %s", msg.content.toString());
                        if (callback) {
                            callback(JSON.parse(msg.content.toString()), msg.fields.routingKey)
                        }
                        if (conf.consumer.noAck === false) {
                            ch.ack(msg);
                        }
                    }, conf.consumer);
                });
            });
        }
    };
};

module.exports = Queue;

We also want to emit messages using an exchange

Emiter:

use G\Rabbit\Builder;
$server = [
    'host' => 'localhost',
    'port' => 5672,
    'user' => 'guest',
    'pass' => 'guest',
];
$exchange = Builder::exchange('process.log', $server);
$exchange->emit("xxx.log", "aaaa");
$exchange->emit("xxx.log", ["11", "aaaa"]);
$exchange->emit("yyy.log", "aaaa");
var builder = require('../../src/Builder');

var server = {
    host: 'localhost',
    port: 5672,
    user: 'guest',
    pass: 'guest'
};

var exchange = builder.exchange('process.log', server);

exchange.emit("xxx.log", "aaaa");
exchange.emit("xxx.log", ["11", "aaaa"]);
exchange.emit("yyy.log", "aaaa");

and receiver:

use G\Rabbit\Builder;
$server = [
    'host' => 'localhost',
    'port' => 5672,
    'user' => 'guest',
    'pass' => 'guest',
];
Builder::exchange('process.log', $server)->receive("yyy.log", function ($routingKey, $data) {
    error_log($routingKey." - ".json_encode($data));
});
var server = {
    host: 'localhost',
    port: 5672,
    user: 'guest',
    pass: 'guest'
};

var exchange = builder.exchange('process.log', server);

exchange.receive("yyy.log", function (routingKey, data) {
    console.log(routingKey, data);
});

And that’s the PHP implementation:

namespace G\Rabbit;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class Exchange
{
    private $name;
    private $conf;
    public function __construct($name, $conf)
    {
        $this->name = $name;
        $this->conf = $conf;
    }
    private function createConnection()
    {
        $server = $this->conf['server'];
        return new AMQPStreamConnection($server['host'], $server['port'], $server['user'], $server['pass']);
    }
    public function emit($routingKey, $data = null)
    {
        $connection = $this->createConnection();
        $channel = $connection->channel();
        $conf = $this->conf['exchange'];
        $channel->exchange_declare($this->name, 'topic', $conf['passive'], $conf['durable'], $conf['auto_delete'],
            $conf['internal'], $conf['nowait']);
        $msg = new AMQPMessage(json_encode($data), [
            'delivery_mode' => 2, # make message persistent
        ]);
        $channel->basic_publish($msg, $this->name, $routingKey);
        $channel->close();
        $connection->close();
    }
    public function receive($bindingKey, callable $callback)
    {
        $connection = $this->createConnection();
        $channel = $connection->channel();
        $conf = $this->conf['exchange'];
        $channel->exchange_declare($this->name, 'topic', $conf['passive'], $conf['durable'], $conf['auto_delete'],
            $conf['internal'], $conf['nowait']);
        $queueConf = $this->conf['queue'];
        list($queue_name, ,) = $channel->queue_declare("", $queueConf['passive'], $queueConf['durable'],
            $queueConf['exclusive'], $queueConf['auto_delete'], $queueConf['nowait']);
        $channel->queue_bind($queue_name, $this->name, $bindingKey);
        $consumerConf = $this->conf['consumer'];
        $channel->basic_consume($queue_name, '', $consumerConf['no_local'], $consumerConf['no_ack'],
            $consumerConf['exclusive'], $consumerConf['nowait'],
            function ($msg) use ($callback) {
                call_user_func($callback, $msg->delivery_info['routing_key'], json_decode($msg->body, true));
                $now = new \DateTime();
                echo '['.$now->format('d/m/Y H:i:s').'] '.$this->name.':'.$msg->delivery_info['routing_key'].'::', $msg->body, "\n";
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            });
        $now = new \DateTime();
        echo '['.$now->format('d/m/Y H:i:s')."] Exchange '{$this->name}' initialized \n";
        while (count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }
}

And node:

var amqp = require('amqplib/callback_api');

var Exchange = function (name, conf) {
    return {
        emit: function (routingKey, data, close = true) {
            amqp.connect(`amqp://${conf.server.user}:${conf.server.pass}@${conf.server.host}:${conf.server.port}`, function (err, conn) {
                conn.createChannel(function (err, ch) {
                    var msg = JSON.stringify(data);
                    ch.assertExchange(name, 'topic', conf.exchange);
                    ch.publish(name, routingKey, new Buffer(msg));
                });
                if (close) {
                    setTimeout(function () {
                        conn.close();
                        process.exit(0)
                    }, 500);
                }
            });
        },
        receive: function (bindingKey, callback) {
            amqp.connect(`amqp://${conf.server.user}:${conf.server.pass}@${conf.server.host}:${conf.server.port}`, function (err, conn) {
                conn.createChannel(function (err, ch) {
                    ch.assertExchange(name, 'topic', conf.exchange);
                    console.log(new Date().toString() + ' Exchange ' + name + ' initialized');
                    ch.assertQueue('', conf.queue, function (err, q) {

                        ch.bindQueue(q.queue, name, bindingKey);

                        ch.consume(q.queue, function (msg) {
                            console.log(new Date().toString(), name, ":", msg.fields.routingKey, "::", msg.content.toString());
                            if (callback) {
                                callback(msg.fields.routingKey, JSON.parse(msg.content.toString()))
                            }
                            if (conf.consumer.noAck === false) {
                                ch.ack(msg);
                            }
                        }, conf.consumer);
                    });
                });
            });
        }
    };
};

module.exports = Exchange;

Finally we want to use RPC commands. In fact RPC implementations is similar than Queue but in this case client will receive an answer.

Client side

use G\Rabbit\Builder;
$server = [
    'host' => 'localhost',
    'port' => 5672,
    'user' => 'guest',
    'pass' => 'guest',
];
echo Builder::rpc('rpc.hello', $server)->call("Gonzalo", "Ayuso");
var builder = require('../../src/Builder');

var server = {
    host: 'localhost',
    port: 5672,
    user: 'guest',
    pass: 'guest'
};

var rpc = builder.rpc('rpc.hello', server);
rpc.call("Gonzalo", "Ayuso", function (data) {
    console.log(data);
});

Server side:

use G\Rabbit\Builder;
$server = [
    'host' => 'localhost',
    'port' => 5672,
    'user' => 'guest',
    'pass' => 'guest',
];
Builder::rpc('rpc.hello', $server)->server(function ($name, $surname) use ($server) {
    return "Hello {$name} {$surname}";
});
var builder = require('../../src/Builder');

var server = {
    host: 'localhost',
    port: 5672,
    user: 'guest',
    pass: 'guest'
};

var rpc = builder.rpc('rpc.hello', server);

rpc.server(function (name, surname) {
    return "Hello " + name + " " + surname;
});

And Implementations:

namespace G\Rabbit;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class RPC
{
    private $name;
    private $conf;
    public function __construct($name, $conf)
    {
        $this->name = $name;
        $this->conf = $conf;
    }
    private function createConnection()
    {
        $server = $this->conf['server'];
        return new AMQPStreamConnection($server['host'], $server['port'], $server['user'], $server['pass']);
    }
    public function call()
    {
        $params = (array)func_get_args();
        $response = null;
        $corr_id = uniqid();
        $connection = $this->createConnection();
        $channel = $connection->channel();
        $queueConf = $this->conf['queue'];
        list($callback_queue, ,) = $channel->queue_declare("", $queueConf['passive'], $queueConf['durable'],
            $queueConf['exclusive'], $queueConf['auto_delete'], $queueConf['nowait']);
        $consumerConf = $this->conf['consumer'];
        $channel->basic_consume($callback_queue, '', $consumerConf['no_local'], $consumerConf['no_ack'],
            $consumerConf['exclusive'], $consumerConf['nowait'], function ($rep) use (&$corr_id, &$response) {
                if ($rep->get('correlation_id') == $corr_id) {
                    $response = $rep->body;
                }
            });
        $msg = new AMQPMessage(json_encode($params), [
            'correlation_id' => $corr_id,
            'reply_to'       => $callback_queue,
        ]);
        $channel->basic_publish($msg, '', $this->name);
        while (!$response) {
            $channel->wait();
        }
        return json_decode($response, true);
    }
    public function server(callable $callback)
    {
        $connection = $this->createConnection();
        $channel = $connection->channel();
        $queueConf = $this->conf['queue'];
        $channel->queue_declare($this->name, $queueConf['passive'], $queueConf['durable'], $queueConf['exclusive'],
            $queueConf['auto_delete'], $queueConf['nowait']);
        $now = new \DateTime();
        echo '['.$now->format('d/m/Y H:i:s')."] RPC server '{$this->name}' initialized \n";
        $channel->basic_qos(null, 1, null);
        $consumerConf = $this->conf['consumer'];
        $channel->basic_consume($this->name, '', $consumerConf['no_local'], $consumerConf['no_ack'],
            $consumerConf['exclusive'],
            $consumerConf['nowait'], function ($req) use ($callback) {
                $response = json_encode(call_user_func_array($callback, array_values(json_decode($req->body, true))));
                $msg = new AMQPMessage($response, [
                    'correlation_id' => $req->get('correlation_id'),
                    'delivery_mode'  => 2, # make message persistent
                ]);
                $req->delivery_info['channel']->basic_publish($msg, '', $req->get('reply_to'));
                $req->delivery_info['channel']->basic_ack($req->delivery_info['delivery_tag']);
                $now = new \DateTime();
                echo '['.$now->format('d/m/Y H:i:s').'] '.$this->name.":: req => '{$req->body}' response=> '{$response}'\n";
            });
        while (count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }
}
var amqp = require('amqplib/callback_api');

var RPC = function (name, conf) {
    var generateUuid = function () {
        return Math.random().toString() +
            Math.random().toString() +
            Math.random().toString();
    };

    return {
        call: function () {
            var params = [];
            for (i = 0; i < arguments.length - 1; i++) {
                params.push(arguments[i]);
            }
            var callback = arguments[arguments.length - 1];

            amqp.connect(`amqp://${conf.server.user}:${conf.server.pass}@${conf.server.host}:${conf.server.port}`, function (err, conn) {
                conn.createChannel(function (err, ch) {
                    ch.assertQueue('', conf.queue, function (err, q) {
                        var corr = generateUuid();

                        ch.consume(q.queue, function (msg) {
                            if (msg.properties.correlationId == corr) {
                                callback(JSON.parse(msg.content.toString()));
                                setTimeout(function () {
                                    conn.close();
                                    process.exit(0)
                                }, 500);
                            }
                        }, conf.consumer);
                        ch.sendToQueue(name,
                            new Buffer(JSON.stringify(params)),
                            {correlationId: corr, replyTo: q.queue});
                    });
                });
            });
        },
        server: function (callback) {
            amqp.connect(`amqp://${conf.server.user}:${conf.server.pass}@${conf.server.host}:${conf.server.port}`, function (err, conn) {
                conn.createChannel(function (err, ch) {
                    ch.assertQueue(name, conf.queue);
                    console.log(new Date().toString() + ' RPC ' + name + ' initialized');
                    ch.prefetch(1);
                    ch.consume(name, function reply(msg) {
                        console.log(new Date().toString(), msg.fields.routingKey, " :: ", msg.content.toString());
                        var response = JSON.stringify(callback.apply(this, JSON.parse(msg.content.toString())));
                        ch.sendToQueue(msg.properties.replyTo,
                            new Buffer(response),
                            {correlationId: msg.properties.correlationId});

                        ch.ack(msg);

                    }, conf.consumer);
                });
            });
        }
    };
};

module.exports = RPC;

You can see whole projects at github: RabbitMQ-php, RabbitMQ-node

Playing with Docker, Silex, Python, Node and WebSockets

I’m learning Docker. In this post I want to share a little experiment that I have done. I know the code looks like over-engineering but it’s just an excuse to build something with docker and containers. Let me explain it a little bit.

The idea is build a Time clock in the browser. Something like this:

Clock

Yes I know. We can do it only with js, css and html but we want to hack a little bit more. The idea is to create:

  • A Silex/PHP frontend
  • A WebSocket server with socket.io/node
  • A Python script to obtain the current time

WebSocket server will open 2 ports: One port to serve webSockets (socket.io) and another one as a http server (express). Python script will get the current time and it’ll send it to the webSocket server. Finally one frontend(silex) will be listening to WebSocket’s event and it will render the current time.

That’s the WebSocket server (with socket.io and express)

var
    express = require('express'),
    expressApp = express(),
    server = require('http').Server(expressApp),
    io = require('socket.io')(server, {origins: 'localhost:*'})
    ;

expressApp.get('/tic', function (req, res) {
    io.sockets.emit('time', req.query.time);
    res.json('OK');
});

expressApp.listen(6400, '0.0.0.0');

server.listen(8080);

That’s our Python script

from time import gmtime, strftime, sleep
import httplib2

h = httplib2.Http()
while True:
    (resp, content) = h.request("http://node:6400/tic?time=" + strftime("%H:%M:%S", gmtime()))
    sleep(1)

And our Silex frontend

use Silex\Application;
use Silex\Provider\TwigServiceProvider;

$app = new Application(['debug' => true]);
$app->register(new TwigServiceProvider(), [
    'twig.path' => __DIR__ . '/../views',
]);

$app->get("/", function (Application $app) {
    return $app['twig']->render('index.twig', []);
});

$app->run();

using this twig template

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Docker example</title>
    <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" integrity="sha384-BVYiiSIFeK1dGmJRAkycuHAHRg32OmUcww7on3RYdg4Va+PmSTsz/K68vbdEjh4u" crossorigin="anonymous">
    <link href="css/app.css" rel="stylesheet">
    <script src="https://oss.maxcdn.com/html5shiv/3.7.3/html5shiv.min.js"></script>
    <script src="https://oss.maxcdn.com/respond/1.4.2/respond.min.js"></script>
</head>
<body>
<div class="site-wrapper">
    <div class="site-wrapper-inner">
        <div class="cover-container">
            <div class="inner cover">
                <h1 class="cover-heading">
                    <div id="display">
                        display
                    </div>
                </h1>
            </div>
        </div>
    </div>
</div>
<script src="//localhost:8080/socket.io/socket.io.js"></script>
<script src="https://ajax.googleapis.com/ajax/libs/jquery/1.12.4/jquery.min.js"></script>
<script>
var socket = io.connect('//localhost:8080');

$(function () {
    socket.on('time', function (data) {
        $('#display').html(data);
    });
});
</script>
</body>
</html>

The idea is to use one Docker container for each process. I like to have all the code in one place so all containers will share the same volume with source code.

First the node container (WebSocket server)

FROM node:argon

RUN mkdir -p /mnt/src
WORKDIR /mnt/src/node

EXPOSE 8080 6400

Now the python container

FROM python:2

RUN pip install httplib2

RUN mkdir -p /mnt/src
WORKDIR /mnt/src/python

And finally Frontend contailer (apache2 with Ubuntu 16.04)

FROM ubuntu:16.04

RUN locale-gen es_ES.UTF-8
RUN update-locale LANG=es_ES.UTF-8
ENV DEBIAN_FRONTEND=noninteractive

RUN apt-get update -y
RUN apt-get install --no-install-recommends -y apache2 php libapache2-mod-php
RUN apt-get clean -y

COPY ./apache2/sites-available/000-default.conf /etc/apache2/sites-available/000-default.conf

RUN mkdir -p /mnt/src

RUN a2enmod rewrite
RUN a2enmod proxy
RUN a2enmod mpm_prefork

RUN chown -R www-data:www-data /mnt/src
ENV APACHE_RUN_USER www-data
ENV APACHE_RUN_GROUP www-data
ENV APACHE_LOG_DIR /var/log/apache2
ENV APACHE_LOCK_DIR /var/lock/apache2
ENV APACHE_PID_FILE /var/run/apache2/apache2.pid
ENV APACHE_SERVERADMIN admin@localhost
ENV APACHE_SERVERNAME localhost

EXPOSE 80

Now we’ve got the three containers but we want to use all together. We’ll use a docker-compose.yml file. The web container will expose port 80 and node container 8080. Node container also opens 6400 but this port is an internal port. We don’t need to access to this port outside. Only Python container needs to access to this port. Because of that 6400 is not mapped to any port in docker-compose

version: '2'

services:
  web:
    image: gonzalo123/example_web
    container_name: example_web
    ports:
     - "80:80"
    restart: always
    depends_on:
      - node
    build:
      context: ./images/php
      dockerfile: Dockerfile
    entrypoint:
      - /usr/sbin/apache2
      - -D
      - FOREGROUND
    volumes:
     - ./src:/mnt/src

  node:
    image: gonzalo123/example_node
    container_name: example_node
    ports:
     - "8080:8080"
    restart: always
    build:
      context: ./images/node
      dockerfile: Dockerfile
    entrypoint:
      - npm
      - start
    volumes:
     - ./src:/mnt/src

  python:
      image: gonzalo123/example_python
      container_name: example_python
      restart: always
      depends_on:
        - node
      build:
        context: ./images/python
        dockerfile: Dockerfile
      entrypoint:
        - python
        - tic.py
      volumes:
       - ./src:/mnt/src

And that’s all. We only need to start our containers

docker-compose up --build -d

and open our browser at: http://localhost to see our Time clock

Full source code available within my github account

Playing with arduino, IoT, crossbar and websockets

Yes. Finally I’ve got an arduino board. It’s time to hack a little bit. Today I want to try different things. I want to display in a webpage one value from my arduino board. For example one analog data using a potentiometer. Let’s start.

We are going to use one potentiometer. A potentiometer is a resistor with a rotating contact that forms an adjustable voltage divider. It has three pins. If we connect one pin to 5V power source of our arduino, another one to the ground and another to one A0 (analog input 0), we can read different values depending on the position of potentiometer’s rotating contact.

arduino_analog

Arduino has 10 bit analog resolution. That means 1024 possible values, from 0 to 1023. So when our potentiometer gives us 5 volts we’ll obtain 1024 and when our it gives us 0V we’ll read 0. Here we can see a simple arduino program to read this analog input and send data via serial port:

int mem;

void setup() {
  Serial.begin(9600);
}

void loop() {
  int value = analogRead(A0);
  if (value != mem) {
    Serial.println(value);
  }
  mem = value;

  delay(100);
}

This program is simple loop with a delay of 100 milliseconds that reads A0 and if value is different than previously read (to avoid sending the same value when nobody is touching the potentiometer) we send the value via serial port (with 9600 bauds)

We can test our program using the serial monitor of our arduino IDE our using another serial monitor.

Now we’re going to create one script to read this serial port data. We’re going to use Python. I’ll use my laptop and my serial port is /dev/tty.usbmodem14231

import serial

arduino = serial.Serial('/dev/tty.usbmodem14231', 9600)

while 1:
  print arduino.readline().strip()

Basically we’ve got our backend running. Now we can create a simple frontend.

...
<div id='display'></div>
...

We’ll need websockets. I normally use socket.io but today I’ll use Crossbar.io. Since I hear about it in a Ronny’s talk at deSymfony conference I wanted to use it.

I’ll change a little bit our backend to emit one event

import serial
from os import environ
from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import LoopingCall
from autobahn.twisted.wamp import ApplicationSession, ApplicationRunner

arduino = serial.Serial('/dev/tty.usbmodem14231', 9600)

class SeriaReader(ApplicationSession):
    @inlineCallbacks
    def onJoin(self, details):
        def publish():
            return self.publish(u'iot.serial.reader', arduino.readline().strip())

        yield LoopingCall(publish).start(0.1)

if __name__ == '__main__':
    runner = ApplicationRunner(environ.get("GONZALO_ROUTER", u"ws://127.0.0.1:8080/ws"), u"iot")
    runner.run(SeriaReader)

Now I only need to create a crossbar.io server. I will use node to do it

var autobahn = require('autobahn'),
    connection = new autobahn.Connection({
            url: 'ws://0.0.0.0:8080/ws',
            realm: 'iot'
        }
    );

connection.open();

And now we only need to connect our frontend to the websocket server

$(function () {
    var connection = new autobahn.Connection({
        url: "ws://192.168.1.104:8080/ws",
        realm: "iot"
    });

    connection.onopen = function (session) {
        session.subscribe('iot.serial.reader', function (args) {
            $('#display').html(args[0]);
        });
    };

    connection.open();
});

It works but thre’s a problem. The first time we connect with our browser we won’t see the display value until we change the position of the potentiometer. That’s because ‘iot.serial.reader’ event is only emitted when potentiometer changes. No change means no new value. To solve this problem we only need to change a little bit our crossbar.io server. We’ll “memorize” the last value and we’ll expose one method ‘iot.serial.get’ to ask about this value

var autobahn = require('autobahn'),
    connection = new autobahn.Connection({
            url: 'ws://0.0.0.0:8080/ws',
            realm: 'iot'
        }
    ),
    mem;

connection.onopen = function (session) {
    session.register('iot.serial.get', function () {
        return mem;
    });

    session.subscribe('iot.serial.reader', function (args) {
        mem = args[0];
    });
};

connection.open();

An now in the frontend we ask for ‘iot.serial.get’ when we connect to the socket

$(function () {
    var connection = new autobahn.Connection({
        url: "ws://192.168.1.104:8080/ws",
        realm: "iot"
    });

    connection.onopen = function (session) {
        session.subscribe('iot.serial.reader', function (args) {
            $('#display').html(args[0]);
        }).then(function () {
                session.call('iot.serial.get').then(
                    function (result) {
                        $('#display').htmlresult);
                    }
                );
            }
        );
    };
    connection.open();
});

And thats all. The source code is available in my github account. You also can see a demo of the working prototype here

Smart bulb controlled from a Raspberry Pi with Python. My RGB alarm clock

I’ve got a BeeWi Smart LED Color Bulb. I must admit I cannot resist to buy those kind of devices :).

I can switch on/off the bulb and change the color using its Mobile App, but it’s not fun. I want to play a little bit with the bulb. My idea is the following one: First switch on the bulb in the mornint and set up the bulb color (Blue for example). Then change bulb color depending on my morning routine. And finally switch the bulb off. Now with this bulb’s color I know if my morning routine is on-time, just looking at the bulb’s color. For example if the bulb is red and I’m still having breakfast probably I’m late.

The prototype is very simple. The bulb has a bluetooth interface and I’ve found a python script to control the bulb. I’ve changed a little bit this script to adapt it to my needs.

Now I only need to set up the crontab within my Raspberry Pi to trigger the script and switch on/off the bulb and change the RGB color.

for example:

# switch on the bulb
/usr/bin/python /mnt/media/projects/iot/bulb.py /mnt/media/projects/iot/conf.json on
# set bulb's color to green
/usr/bin/python /mnt/media/projects/iot/bulb.py /mnt/media/projects/iot/conf.json colour 999900

In another post we play with Telegram bots to read temperature. Now I’ve adapted also my bot to switch on/off and change color of the bulb.

Now I’ve got another toy in my desk. One arduino board. I’m sure I will enjoy a lot 🙂

Performing UPSERT (Update or Insert) with PostgreSQL and PHP

That’s a typical situation. Imagine you’ve got one table

CREATE TABLE PUBLIC.TBUPSERTEXAMPLE
(
  KEY1 CHARACTER VARYING(10) NOT NULL,
  KEY2 CHARACTER VARYING(14) NOT NULL,
  KEY3 CHARACTER VARYING(14) NOT NULL,
  KEY4 CHARACTER VARYING(14) NOT NULL,

  VALUE1 CHARACTER VARYING(20),
  VALUE2 CHARACTER VARYING(20) NOT NULL,
  VALUE3 CHARACTER VARYING(100),
  VALUE4 CHARACTER VARYING(400),
  VALUE5 CHARACTER VARYING(20),

  CONSTRAINT TBUPSERTEXAMPLE_PKEY PRIMARY KEY (KEY1, KEY2, KEY3, KEY4)
)

And you need to update one record. You can perform a simple UPDATE statement but what happens the first time?

You cannot update the record basically because the record doesn’t exists. You need to create an INSERT statement instead. We can do it following different ways. You can create first a SELECT statement and, if the record exists, perform an UPDATE. If it doesn’t exists you perform an INSERT. We also can perform an UPDATE and see how many records are affected. If no records are affected then we perform an INSERT. Finally we can perform one INSERT and it it throws an error then perform an UPDATE.

All of these techniques works in one way or another but PostgreSQL gives us one cool way of doing this operation with one SQL sentence. We can use CTE (Common Table Expression) and execute something like this:

WITH upsert AS (
    UPDATE PUBLIC.TBUPSERTEXAMPLE
    SET
        VALUE1 = :VALUE1,
        VALUE2 = :VALUE2,
        VALUE3 = :VALUE3,
        VALUE4 = :VALUE4,
        VALUE5 = :VALUE5
    WHERE
        KEY1 = :KEY1 AND
        KEY2 = :KEY2 AND
        KEY3 = :KEY3 AND
        KEY4 = :KEY4
    RETURNING *
)
INSERT INTO PUBLIC.TBUPSERTEXAMPLE(KEY1, KEY2, KEY3, KEY4, VALUE1, VALUE2, VALUE3, VALUE4, VALUE5)
SELECT
    :KEY1, :KEY2, :KEY3, :KEY4, :VALUE1, :VALUE2, :VALUE3, :VALUE4, :VALUE5
WHERE
    NOT EXISTS (SELECT 1 FROM upsert);

Since PostgreSQL 9.5 we also can do another technique to do this UPSERT operations. We can do something like this:

INSERT INTO PUBLIC.TBUPSERTEXAMPLE (key1, key2, key3, key4, value1, value2, value3, value4, value5)
  VALUES ('key2', 'key2', 'key3', 'key4', 'value1',  'value2',  'value3',  'value4',  'value5')
ON CONFLICT (key1, key2, key3, key4)
DO UPDATE SET 
  value1 = 'value1', 
  value2 = 'value2', 
  value3 = 'value3', 
  value4 = 'value4', 
  value5 = 'value5'
WHERE 
  TBUPSERTEXAMPLE.key1 = 'key2' AND 
  TBUPSERTEXAMPLE.key2 = 'key2' AND 
  TBUPSERTEXAMPLE.key3 = 'key3' AND 
  TBUPSERTEXAMPLE.key4 = 'key4';

To help me writing this sentence I’ve created a simple PHP wrapper:

Here one example using PDO

use G\SqlUtils\Upsert;

$conn = new PDO('pgsql:dbname=gonzalo;host=localhost', 'username', 'password');
$conn->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

$conn->beginTransaction();
try {
    Upsert::createFromPDO($conn)->exec('PUBLIC.TBUPSERTEXAMPLE', [
        'KEY1' => 'key1',
        'KEY2' => 'key2',
        'KEY3' => 'key3',
        'KEY4' => 'key4',
    ], [
        'VALUE1' => 'value1',
        'VALUE2' => 'value2',
        'VALUE3' => 'value3',
        'VALUE4' => 'value4',
        'VALUE5' => 'value5',
    ]);
    $conn->commit();
} catch (Exception $e) {
    $conn->rollback();
    throw $e;
}

And another one using DBAL

use Doctrine\DBAL\DriverManager;
use G\SqlUtils\Upsert;

$connectionParams = [
    'dbname'   => 'gonzalo',
    'user'     => 'username',
    'password' => 'password',
    'host'     => 'localhost',
    'driver'   => 'pdo_pgsql',
];

$dbh = DriverManager::getConnection($connectionParams);
$dbh->transactional(function ($conn) {
    Upsert::createFromDBAL($conn)->exec('PUBLIC.TBUPSERTEXAMPLE', [
        'KEY1' => 'key1',
        'KEY2' => 'key2',
        'KEY3' => 'key3',
        'KEY4' => 'key4',
    ], [
        'VALUE1' => 'value1',
        'VALUE2' => 'value2',
        'VALUE3' => 'value3',
        'VALUE4' => null,
        'VALUE5' => 'value5',
    ]);
});

And that’s all. Library is available in my github and it’s also at packagist.

Encrypt Websocket (socket.io) communications

I’m a big fan of WebSockets and socket.io. I’ve written a lot of about it. In last posts I’ve written about socket.io and authentication. Today we’re going to speak about communications.

Imagine we’ve got a websocket server and we connect our application to this server (even using https/wss). If we open our browser’s console we can inspect our WebSocket communications. We also can enable debugging. This works in a similar way than when we start the promiscuous mode within our network interface. We will see every packets. Not only the packets that server is sending to us.

If we send send sensitive information over websockets, that means than one logged user can see another ones information. We can separate namespaces in our socket.io server. We also can do another thing: Encrypt communications using crypto-js.

I’ve created one small wrapper to use it with socket.io.
We can install our server dependency

npm g-crypt

And install our client dependency with bower

bower install g-crypt

And use it in our server

var io = require('socket.io')(3000),
    Crypt = require("g-crypt"),
    passphrase = 'super-secret-passphrase',
    crypter = Crypt(passphrase);

io.on('connection', function (socket) {
    socket.on('counter', function (data) {
        var decriptedData = crypter.decrypt(data);
        setTimeout(function () {
            console.log("counter status: " + decriptedData.id);
            decriptedData.id++;
            socket.emit('counter', crypter.encrypt(decriptedData));
        }, 1000);
    });
});

And now a simple HTTP application

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
Open console to see the messages

<script src="http://localhost:3000/socket.io/socket.io.js"></script>
<script src="assets/cryptojslib/rollups/aes.js"></script>
<script src="assets/g-crypt/src/Crypt.js"></script>
<script>
    var socket = io('http://localhost:3000/'),
        passphrase = 'super-secret-passphrase',
        crypter = Crypt(passphrase),
        id = 0;

    socket.on('connect', function () {
        console.log("connected! Let's start the counter with: " + id);
        socket.emit('counter', crypter.encrypt({id: id}));
    });

    socket.on('counter', function (data) {
        var decriptedData = crypter.decrypt(data);
        console.log("counter status: " + decriptedData.id);
        socket.emit('counter', crypter.encrypt({id: decriptedData.id}));
    });
</script>

</body>
</html>

Now our communications are encrypted and logged user cannot read another ones data.

Library is a simple wrapper

Crypt = function (passphrase) {
    "use strict";
    var pass = passphrase;
    var CryptoJSAesJson = {
        parse: function (jsonStr) {
            var j = JSON.parse(jsonStr);
            var cipherParams = CryptoJS.lib.CipherParams.create({ciphertext: CryptoJS.enc.Base64.parse(j.ct)});
            if (j.iv) cipherParams.iv = CryptoJS.enc.Hex.parse(j.iv);
            if (j.s) cipherParams.salt = CryptoJS.enc.Hex.parse(j.s);
            return cipherParams;
        },
        stringify: function (cipherParams) {
            var j = {ct: cipherParams.ciphertext.toString(CryptoJS.enc.Base64)};
            if (cipherParams.iv) j.iv = cipherParams.iv.toString();
            if (cipherParams.salt) j.s = cipherParams.salt.toString();
            return JSON.stringify(j);
        }
    };

    return {
        decrypt: function (data) {
            return JSON.parse(CryptoJS.AES.decrypt(data, pass, {format: CryptoJSAesJson}).toString(CryptoJS.enc.Utf8));
        },
        encrypt: function (data) {
            return CryptoJS.AES.encrypt(JSON.stringify(data), pass, {format: CryptoJSAesJson}).toString();
        }
    };
};

if (typeof module !== 'undefined' && typeof module.exports !== 'undefined') {
    CryptoJS = require("crypto-js");
    module.exports = Crypt;
} else {
    window.Crypt = Crypt;
}

Library available in my github and also we can use it using npm and bower.

Home automation pet project. Playing with IoT, temperature sensors, fans and Telegram bots

Summer holidays are over. Besides my bush walks I’ve been also hacking a little bit with one idea that I had in mind. Summer means high temperatures and I wanted to control my fan. For example turn on the fan when temperature is over a threshold. I can do it using an Arduino board and a temperature sensor, but I don’t have the one Arduino board. I have several devices. For example a Wemo switch. With this device connected to my Wifi network I can switch on and off my fan remotely from my mobile phone (using its android app) or even from my Pebble watch using the API. I also have a BeeWi temperature/humidity sensor. It’s a BTLE device. It comes with its own app for android, but there’s also a API. Yes. I known that one Arduino board with a couple of sensors can be cheaper than one of this devices, but when I’m a shop and I’ve got one of this devices in my hands I cannot resist.

I also have a new Raspberry pi 3. I’ve recently upgraded my home multimedia server from a rpi2 to the new rpi3. Basically I use it as multimedia server and now also as retro console. This new rpi3 has Bluetooth so I wanted to do something with it. Read temperature from the Bluetooth sensor sounds good so I started to hack a little bit.

I found this post. I started working with Python. The script almost works but it uses Bluetooth connection and as someone said in the comments it uses a lot of battery. So I switched to a BTLE version. I found a simple node library to connect BTLE devices called noble, really simple to use. In one afternoon I had one small script ready. The idea was put this script in my RP3’s crontab, and scan the temperature each minute (via noble) and if the temperature was over a threshold switch on the wemo device (via ouimeaux). I also wanted to be informed when my fan is switch on and off. The most easier way to do it was via Telegram (I already knew telebot library).

var noble = require('noble'),
    Wemo = require('wemo-client'),
    TeleBot = require('telebot'),
    fs = require('fs'),
    beeWiData,
    wemo,
    threshold,
    address,
    bot,
    chatId,
    wemoDevice,
    configuration,
    confPath;

if (process.argv.length <= 2) {
    console.log("Usage: " + __filename + " conf.json");
    process.exit(-1);
}

confPath = process.argv[2];
try {
    configuration = JSON.parse(
        fs.readFileSync(process.argv[2])
    );
} catch (e) {
    console.log("configuration file not valid");
    process.exit(-1);
}

bot = new TeleBot(configuration.telegramBotAPIKey);
address = configuration.beeWiAddress;
threshold = configuration.threshold;
wemoDevice = configuration.wemoDevice;
chatId = configuration.telegramChatId;

function persists() {
    configuration.beeWiData = beeWiData;
    fs.writeFileSync(confPath, JSON.stringify(configuration));
}

function setSwitchState(state, callback) {
    wemo = new Wemo();
    wemo.discover(function(deviceInfo) {
        if (deviceInfo.friendlyName == wemoDevice) {
            console.log("device found:", deviceInfo.friendlyName, "setting the state to", state);
            var client = wemo.client(deviceInfo);
            client.on('binaryState', function(value) {
                callback();
            });

            client.on('statusChange', function(a) {
                console.log("statusChange", a);
            });
            client.setBinaryState(state);
        }
    });
}

beeWiData = {temperature: undefined, humidity: undefined, batery: undefined};

function hexToInt(hex) {
    if (hex.length % 2 !== 0) {
        hex = "0" + hex;
    }
    var num = parseInt(hex, 16);
    var maxVal = Math.pow(2, hex.length / 2 * 8);
    if (num > maxVal / 2 - 1) {
        num = num - maxVal;
    }
    return num;
}

noble.on('stateChange', function(state) {
    if (state === 'poweredOn') {
        noble.stopScanning();
        noble.startScanning();
    } else {
        noble.stopScanning();
    }
});

noble.on('scanStop', function() {
    var message, state;
    if (beeWiData.temperature > threshold) {
        state = 1;
        message = "temperature (" + beeWiData.temperature + ") over threshold (" + threshold + "). Fan ON. Humidity: " + beeWiData.humidity;
    } else {
        message = "temperature (" + beeWiData.temperature + ") under threshold (" + threshold + "). Fan OFF. Humidity: " + beeWiData.humidity;
        state = 0;
    }
    setSwitchState(state, function() {
        if (configuration.beeWiData.hasOwnProperty('temperature') && configuration.beeWiData.temperature < threshold && state === 1 || configuration.beeWiData.temperature > threshold && state === 0) {
            console.log("Notify to telegram bot", message);
            bot.sendMessage(chatId, message).then(function() {
                process.exit(0);
            }, function(e) {
                console.error(e);
                process.exit(0);
            });
            persists();
        } else {
            console.log(message);
            persists();
            process.exit(0);
        }
    });
});

noble.on('discover', function(peripheral) {
    if (peripheral.address == address) {
        var data = peripheral.advertisement.manufacturerData.toString('hex');
        beeWiData.temperature = parseFloat(hexToInt(data.substr(10, 2)+data.substr(8, 2))/10).toFixed(1);
        beeWiData.humidity = Math.min(100,parseInt(data.substr(14, 2),16));
        beeWiData.batery = parseInt(data.substr(24, 2),16);
        beeWiData.date = new Date();
        noble.stopScanning();
    }
});

setTimeout(function() {
    console.error("timeout exceded!");
    process.exit(0);
}, 5000);

The script is here.

It works but I wanted to keep on hacking. One Sunday morning I read this post. I don’t have an amazon button, but I wanted to do something similar. I started to play with scapy library sniffing ARP packets in my home network. I realize that I can detect when my Kindle connects to the network, my tv, or even my mobile phone. Then I had one I idea: Detect when my mobile phone connects to my wifi. My mobile phone connects to my wifi before I enter in my house so my idea was simple: Detect when I’m close to my home’s door and send me a telegram message saying “Wellcome home” in addition to the temperature inside my house at this moment.

#!/usr/bin/env python

import sys
from scapy.all import *
import telebot
import gearman
import json
from StringIO import StringIO

BUFFER_SIZE = 1024

try:
    with open(sys.argv[1]) as data_file:
        data = json.load(data_file)
        myPhone = data['myPhone']
        routerIP = data['routerIP']
        TOKEN = data['telegramBotAPIKey']
        chatID = data['telegramChatId']
        gearmanServer = data['gearmanServer']
except:
    print("Unexpected error:", sys.exc_info()[0])
    raise

def getSensorData():
    gm_client = gearman.GearmanClient([gearmanServer])
    completed_job_request = gm_client.submit_job("temp", '')
    io = StringIO(completed_job_request.result)

    return json.load(io)

tb = telebot.TeleBot(TOKEN)

def arp_display(pkt):
    if pkt[ARP].op == 1 and pkt[ARP].hwsrc == myPhone and pkt[ARP].pdst == routerIP:
        sensorData = getSensorData()
        message = "Wellcome home Gonzalo! Temperature: %s humidity: %s" % (sensorData['temperature'], sensorData['humidity'])
        tb.send_message(chatID, message)
        print message

print sniff(prn=arp_display, filter='arp', store=0)

I have one node script to read temperature and one Python script to sniff my network. I can find how to read temperature from Python and use only one script but I was lazy (remember that I was on holiday) so I turned the node script that reads temperature into a gearman worker.

var noble = require('noble'),
    fs = require('fs'),
    Gearman = require('node-gearman'),
    beeWiData,
    address,
    bot,
    configuration,
    confPath,
    status,
    callback;

var gearman = new Gearman();

if (process.argv.length <= 2) {
    console.log("Usage: " + __filename + " conf.json");
    process.exit(-1);
}

confPath = process.argv[2];
try {
    configuration = JSON.parse(
        fs.readFileSync(process.argv[2])
    );
} catch (e) {
    console.log("configuration file not valid", e);
    process.exit(-1);
}

address = configuration.beeWiAddress;
delay = configuration.tempServerDelayMinutes * 60 * 1000;
tcpPort = configuration.tempServerPort;

beeWiData = {};

function hexToInt(hex) {
    if (hex.length % 2 !== 0) {
        hex = "0" + hex;
    }
    var num = parseInt(hex, 16);
    var maxVal = Math.pow(2, hex.length / 2 * 8);
    if (num > maxVal / 2 - 1) {
        num = num - maxVal;
    }
    return num;
}

noble.on('stateChange', function(state) {
    if (state === 'poweredOn') {
        console.log("stateChange:poweredOn");
        status = true;
    } else {
        status = false;
    }
});

noble.on('discover', function(peripheral) {
    if (peripheral.address == address) {
        var data = peripheral.advertisement.manufacturerData.toString('hex');
        beeWiData.temperature = parseFloat(hexToInt(data.substr(10, 2)+data.substr(8, 2))/10).toFixed(1);
        beeWiData.humidity = Math.min(100,parseInt(data.substr(14, 2),16));
        beeWiData.batery = parseInt(data.substr(24, 2),16);
        beeWiData.date = new Date();
        noble.stopScanning();
    }
});

noble.on('scanStop', function() {
    console.log(beeWiData);
    noble.stopScanning();
    callback();
});

var worker;

function workerCallback(payload, worker) {
    callback = function() {
        worker.end(JSON.stringify(beeWiData));
    }

    beeWiData = {temperature: undefined, humidity: undefined, batery: undefined};

    if (status) {
        noble.stopScanning();
        noble.startScanning();
    } else {
        setInterval(function() {
            workerCallback(payload, worker);
        }, 1000);
    }
}

gearman.registerWorker("temp", workerCallback);

Now I only need to call this worker from my Python sniffer and thats all.

I wanted to play a little bit. I also wanted to ask the temperature on demand. Since I was using Telegram I had an idea. Create a Telegram bot running in my RP3. And that’s my summer pet project. Basically it has three parts:

worker.js
It’s a gearman worker. It reads temperature and humidity from my BeeWi sensor via BTLE

bot.py
It’s a Telegram bot with the following commands available:

/switchInfo: get switch info
/switchOFF: switch OFF the switch
/help: Gives you information about the available commands
/temp: Get temperature
/switchON: switch ON the switch

sniff.py
It’s just a ARP sniffer. It detects when I’m close to my home and sends me a message via Telegram with the temperature. It detects when my mobile phone sends a ARP package to my router (aka when I connect to my Wifi). It happens before I enter in my house, so the Telegram message arrives before I put the key in the door 🙂

I run al my scripts in my Raspberry Pi3. To ensure all scripts are up an running I use supervisor

All the scripts are available in my github account

i18n angular2 service

This days I’m learning angular2 and ionic2. My first impression was bad. Too many new things: angular2, typescript, new project architecture, … but after passing those bad moments I started to like angular2. One of the first things that I want to work with is internationalization. There’re several i18n providers for angular2. For example ng2-translate, but I want to build something similar to a angular i18n provider that I’d created time ago.

The idea is first configure the service

import {Component} from "@angular/core";
import {Platform, ionicBootstrap} from "ionic-angular";
import {StatusBar} from "ionic-native";
import {HomePage} from "./pages/home/home";
import {I18nService} from "./services/i18n/I18nService";
import {lang} from "./conf/conf";

@Component({
    template: '<ion-nav [root]="rootPage"></ion-nav>'
})
export class MyApp {
    rootPage:any = HomePage;

    constructor(platform:Platform, i18n:I18nService) {
        platform.ready().then(() => {
            i18n.init(lang);
            StatusBar.styleDefault();
        });
    }
}

ionicBootstrap(MyApp, [I18nService]);

We can see that we’re using one configuration from conf/conf.ts

var lang = {
    availableLangs: {
        en: 'English'
    },
    defaultLang: 'en',
    lang: {
        "true": {
            en: "True",
            es: "Verdadero"
        },
        "false": {
            en: "False",
            es: "Falso"
        },
    }
};

export {lang};

Now we can translate texts using:

import {Component} from "@angular/core";
import {I18nService} from "../../services/i18n/I18nService";

@Component({
    templateUrl: 'build/pages/home/home.html'
})
export class HomePage {
    constructor(private i18n:I18nService) {
    }

    label1:string = this.i18n.translate('true');
    label2:string = 'true';
    
    ngOnInit() {
        setInterval(() => {
            this.label2 = this.i18n.translate(this.label2 == 'false' ? 'true' : 'false');
        }, 2000);
    }
}

We also need a Pipe to use i18n within templates

{{ "true" | i18n }}

This is the service:

import {Injectable, Pipe} from "@angular/core";

@Injectable()
export class I18nService {

    private conf:any;
    private userLang:string;
    private lang:any;

    setUserLang(lang:string):void {
        this.userLang = lang;
    }

    init(lang:any):void {
        this.conf = lang;
        this.setUserLang(lang.defaultLang);
    }

    translate(key:string):string {
        if (typeof this.conf.lang !== 'undefined' && this.conf.lang.hasOwnProperty(key)) {
            return this.conf.lang[key][this.userLang] || key;
        } else {
            return key;
        }
    }
}

@Pipe({
    name: 'i18n',
    pure: false
})
export class I18nPipe {
    constructor(private i18n:I18nService) {
    }

    transform(key:string) {
        return this.i18n.translate(key);
    }
}

The code is in my github account. I also have tried to created a npm package and install it using

npm install angular2-i18n --save

I don’t know why it doesn’t work when I use the npm installed package and I works like a charm when I use the code in my project directory

import {I18nService} from "./services/i18n/I18nService"; // works
import {I18nService} from "angular2-i18n/src/I18nService"; // don't work

I also have problems with unit tests. It works when I run the tests locally (it works on my machine 🙂 ) but it doesn’t work when I run them in travis. Any help with those two problems would be appreciated.

Notify events from PostgreSQL to external listeners

Sometimes we need to call external programs from our PostgreSQL database. We can send sockets from SQL statements. I’ve written about it. The problem with this approach the following one. If user rollbacks the transaction the socket has been already emitted. That’s a problem (or not. Depending on our application). Nobody also guarantees that the process behind the socket server has access to the data of the transaction. If we’re very fast maybe the transaction isn’t commited yet. We can use one sleep function but sleep functions are always a bad idea. PostgreSQL gives us another tool to decouple processes: LISTEN and NOTIFY.

Let me show you and example. First we create a table:

CREATE TABLE PUBLIC.TBLEXAMPLE
(
  KEY1 CHARACTER VARYING(10) NOT NULL,
  KEY2 CHARACTER VARYING(14) NOT NULL,

  VALUE1 CHARACTER VARYING(20),
  VALUE2 CHARACTER VARYING(20) NOT NULL,

  CONSTRAINT TBLEXAMPLE_PKEY PRIMARY KEY (KEY1, KEY2)
)

Now we add a “after insert” trigger to our table

CREATE TRIGGER TBLEXAMPLE_AFTER
AFTER INSERT
ON PUBLIC.TBLEXAMPLE
FOR EACH ROW
EXECUTE PROCEDURE PUBLIC.NOTIFY();

And now, within the trigger function, we send a notify event (‘myEvent’ in this case) with the row information. We need to send plain text in the notify event so we’ll use JSON to encode our row data.

CREATE OR REPLACE FUNCTION PUBLIC.NOTIFY() RETURNS trigger AS
$BODY$
BEGIN
  PERFORM pg_notify('myEvent', row_to_json(NEW)::text);
  RETURN new;
END;
$BODY$
LANGUAGE 'plpgsql' VOLATILE COST 100;

Now we’re going to build a server side example that connects to our PostgreSQL database and listen to the event. In this case we’re going to use nodejs to build the prototype. This example also will enqueue events into a gearman server.

var pg = require('pg'),
    gearmanode = require('gearmanode'),
    gearmanClient,
    conString = 'tcp://username:password@localhost:5432/gonzalo',
    pgClient;

gearmanode.Client.logger.transports.console.level = 'error';

gearmanClient = gearmanode.client();

console.log('LISTEN myEvent');
pgClient = new pg.Client(conString);
pgClient.connect();
pgClient.query('LISTEN myEvent');
pgClient.on('notification', function (data) {
    console.log("\033[34m" + new Date + '-\033[0m payload', data.payload);
    gearmanClient.submitJob('sms.sender.one', data.payload);
});

And that’s all. Now we only need to perform an INSERT statement into our table. This process will trigger our event and our nodejs will enqueue the process into a gearman queue.

INSERT INTO PUBLIC.TBLEXAMPLE(KEY1, KEY2, VALUE1, VALUE2) VALUES ('k1', 'k2', 'v1', 'v2');

It’s good to remark that if our insert statement is inside a transaction and we rollback it, notify won’t send any message.