trunk/services/messagereceiver.php

<?php

include_once('../core/om_debug.inc.php');

(defined("DEBUG") ? "" : define("DEBUG", EXTRA_VERBOSE_MODE));

include_once('../core/om_logger.class.php');

require_once('./php/php-amqplib/vendor/autoload.php');
use PhpAmqpLib\Connection\AMQPConnection;


define('HTTP_RESULT_OK', '200');


function parse($return)
{
    // remplace les '{' et '}' par chaine vide
    $return = str_replace('{', '', $return);
    $return = str_replace('}', '', $return);
    $return = trim($return, " \t");
    // remplace les retours de ligne
    $return = preg_replace('/\s\s+/', '', $return);
    $return = str_replace("\"", "", $return);
    $return = str_replace(": ", ":", $return);
    
    $result = array();
    $resparray = explode(',', $return);

    if ($resparray) {
        foreach ($resparray as $resp) {
            $keyvalue = explode(':', $resp);            
            $result[trim(str_replace('"', '',$keyvalue[0]))] =
                trim(str_replace('"', '', $keyvalue[1]));
        }
    }
    return $result;
}

function postViaCurl($msg) {
    $data = $msg->body;
    
    $cu = curl_init('http://localhost/~mirna/openfoncier/trunk/services/messages');
    curl_setopt($cu, CURLOPT_CUSTOMREQUEST, "POST");
    curl_setopt($cu, CURLOPT_POSTFIELDS, $msg->body);
    curl_setopt($cu, CURLOPT_RETURNTRANSFER, true);
    curl_setopt($cu, CURLOPT_HTTPHEADER, array(                                                                          
        'Content-Type: application/json',
        'Content-Length: ' . strlen($msg->body))
    );

    $result = curl_exec($cu);
    curl_close($cu);

    // le resultat recu par curl est une chaine des caracteres qui contient
    // un tableau
    $result = parse($result);
    
    // log le retour
    logger::instance()->log($result['http_code_message'] . ': ' .
                            $result['message'], EXTRA_VERBOSE_MODE);
    logger::instance()->writeLogToFile();
    
    if ($result['http_code'] == HTTP_RESULT_OK) {
        // accuse de reception du message    
        $msg->delivery_info['channel']->
            basic_ack($msg->delivery_info['delivery_tag']);
            
        // Envoie un message avec la chaine "quit" pour canceler le consumeur.
        if ($msg->body === 'quit') {
            $msg->delivery_info['channel']->
                basic_cancel($msg->delivery_info['consumer_tag']);
        }
    }
}


// pour des besoins de logging il faut setter $_SERVER["REQUEST_URI"]
if (!isset($_SERVER["REQUEST_URI"])) {
    $_SERVER["REQUEST_URI"] = __FILE__;
}


$exchange = 'router';
$queue = 'ERP_CHANNEL';
$consumer_tag = 'consumer';


$connection_host = 'localhost';
$connection_port = '5672';
$connection_user = 'guest';
$connection_password = 'guest';
$connection_vhost = '/';      


$conn = new AMQPConnection($connection_host, $connection_port, $connection_user,
                           $connection_password, $connection_vhost);
                           
$ch = $conn->channel();
if (is_null($ch)) {
    // probleme avec la connexion, donc log le fait
    logger::instance()->log('Probleme dans l\'etablisement de la connexion',
                            EXTRA_VERBOSE_MODE);
    logger::instance()->writeLogToFile();
}

$ch->queue_declare($queue, false, true, false, false);

$ch->exchange_declare($exchange, 'direct', false, true, false);

$ch->queue_bind($queue, $exchange);

$ch->basic_consume($queue, $consumer_tag, false, false, false, false, 'postViaCurl');

function shutdown($ch, $conn) {
    $ch->close();
    $conn->close();
}
register_shutdown_function('shutdown', $ch, $conn);

// Loop as long as the channel has callbacks registered
while (count($ch->callbacks)) {
    $ch->wait();
}


?>
1	<?php
2
3	include_once('../core/om_debug.inc.php');
4
5	(defined("DEBUG") ? "" : define("DEBUG", EXTRA_VERBOSE_MODE));
6
7	include_once('../core/om_logger.class.php');
8
9	require_once('./php/php-amqplib/vendor/autoload.php');
10	use PhpAmqpLib\Connection\AMQPConnection;
11
12
13	define('HTTP_RESULT_OK', '200');
14
15
16	function parse($return)
17	{
18	// remplace les '{' et '}' par chaine vide
19	$return = str_replace('{', '', $return);
20	$return = str_replace('}', '', $return);
21	$return = trim($return, " \t");
22	// remplace les retours de ligne
23	$return = preg_replace('/\s\s+/', '', $return);
24	$return = str_replace("\"", "", $return);
25	$return = str_replace(": ", ":", $return);
26
27	$result = array();
28	$resparray = explode(',', $return);
29
30	if ($resparray) {
31	foreach ($resparray as $resp) {
32	$keyvalue = explode(':', $resp);
33	$result[trim(str_replace('"', '',$keyvalue[0]))] =
34	trim(str_replace('"', '', $keyvalue[1]));
35	}
36	}
37	return $result;
38	}
39
40	function postViaCurl($msg) {
41	$data = $msg->body;
42
43	$cu = curl_init('http://localhost/~mirna/openfoncier/trunk/services/messages');
44	curl_setopt($cu, CURLOPT_CUSTOMREQUEST, "POST");
45	curl_setopt($cu, CURLOPT_POSTFIELDS, $msg->body);
46	curl_setopt($cu, CURLOPT_RETURNTRANSFER, true);
47	curl_setopt($cu, CURLOPT_HTTPHEADER, array(
48	'Content-Type: application/json',
49	'Content-Length: ' . strlen($msg->body))
50	);
51
52	$result = curl_exec($cu);
53	curl_close($cu);
54
55	// le resultat recu par curl est une chaine des caracteres qui contient
56	// un tableau
57	$result = parse($result);
58
59	// log le retour
60	logger::instance()->log($result['http_code_message'] . ': ' .
61	$result['message'], EXTRA_VERBOSE_MODE);
62	logger::instance()->writeLogToFile();
63
64	if ($result['http_code'] == HTTP_RESULT_OK) {
65	// accuse de reception du message
66	$msg->delivery_info['channel']->
67	basic_ack($msg->delivery_info['delivery_tag']);
68
69	// Envoie un message avec la chaine "quit" pour canceler le consumeur.
70	if ($msg->body === 'quit') {
71	$msg->delivery_info['channel']->
72	basic_cancel($msg->delivery_info['consumer_tag']);
73	}
74	}
75	}
76
77
78
79	// pour des besoins de logging il faut setter $_SERVER["REQUEST_URI"]
80	if (!isset($_SERVER["REQUEST_URI"])) {
81	$_SERVER["REQUEST_URI"] = __FILE__;
82	}
83
84
85
86	$exchange = 'router';
87	$queue = 'ERP_CHANNEL';
88	$consumer_tag = 'consumer';
89
90
91	$connection_host = 'localhost';
92	$connection_port = '5672';
93	$connection_user = 'guest';
94	$connection_password = 'guest';
95	$connection_vhost = '/';
96
97
98	$conn = new AMQPConnection($connection_host, $connection_port, $connection_user,
99	$connection_password, $connection_vhost);
100
101	$ch = $conn->channel();
102	if (is_null($ch)) {
103	// probleme avec la connexion, donc log le fait
104	logger::instance()->log('Probleme dans l\'etablisement de la connexion',
105	EXTRA_VERBOSE_MODE);
106	logger::instance()->writeLogToFile();
107	}
108
109	$ch->queue_declare($queue, false, true, false, false);
110
111	$ch->exchange_declare($exchange, 'direct', false, true, false);
112
113	$ch->queue_bind($queue, $exchange);
114
115	$ch->basic_consume($queue, $consumer_tag, false, false, false, false, 'postViaCurl');
116
117	function shutdown($ch, $conn) {
118	$ch->close();
119	$conn->close();
120	}
121	register_shutdown_function('shutdown', $ch, $conn);
122
123	// Loop as long as the channel has callbacks registered
124	while (count($ch->callbacks)) {
125	$ch->wait();
126	}
127
128
129	?>