services/outgoing/messagesender.php

<?php

include_once('../../core/om_debug.inc.php');
include_once('../../dyn/services.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;


// Si le AMQP_DEBUG est defini a true, les messages DEBUG sont envoyes aux
// stdout
if (DEBUG > PRODUCTION_MODE) {
    define('AMQP_DEBUG', true);
}


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;
    
    global $ERP_URL_MESSAGES;
    if (empty($ERP_URL_MESSAGES)) {
        logger::instance()->cleanLog();
        logger::instance()->log("Le URL de destination manque", EXTRA_VERBOSE_MODE);
        logger::instance()->writeLogToFile();
        return;
    }    
    
    $curl_resource = curl_init($ERP_URL_MESSAGES);
    curl_setopt($curl_resource, CURLOPT_CUSTOMREQUEST, "POST");
    curl_setopt($curl_resource, CURLOPT_POSTFIELDS, $msg->body);
    curl_setopt($curl_resource, CURLOPT_RETURNTRANSFER, true);
    curl_setopt($curl_resource, CURLOPT_HTTPHEADER, array(                                                                          
        'Content-Type: application/json',
        'Content-Length: ' . strlen($msg->body))
    );

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


    // le resultat recu par curl est une chaine des caracteres qui contient
    // un tableau
    $result = parse($result);
    
    // log le retour
    logger::instance()->cleanLog();
    logger::instance()->log($result['http_code_message'] . ': ' .
                            $result['message'].' Message: '.
                            $msg->body, 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']);            
    } else {
        sleep(2);
        // accuse de reception du message    
        $msg->delivery_info['channel']->basic_reject(
                            $msg->delivery_info['delivery_tag'], true);
            
    }
    // 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__;
}

// si le $argv[1] manque on ne peut pas cree un tag unique pour le
// consumeur et dans ce cas-la on termine l'execution
$suffix = '';
if (!empty($argv) && count($argv) > 1) {
    $suffix = $argv[1];
} else {
    logger::instance()->log('Missing the command line argument used for tag suffix',
                            EXTRA_VERBOSE_MODE);
    logger::instance()->writeLogToFile();
    exit;
}

$exchange = 'router';
$queue = $ERP_QUEUE;
$consumer_tag = 'consumer'.$suffix;

$conn = new AMQPConnection($ERP_CONNECTION_HOST, $ERP_CONNECTIOn_PORT,
                           $ERP_CONNECTION_USER, $ERP_CONNECTION_PASSWORD,
                           $ERP_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	mlimic	824	<?php
2
3	mlimic	829	include_once('../../core/om_debug.inc.php');
4	mlimic	830	include_once('../../dyn/services.inc.php');
5	mlimic	824
6			(defined("DEBUG") ? "" : define("DEBUG", EXTRA_VERBOSE_MODE));
7
8	mlimic	829	include_once('../../core/om_logger.class.php');
9	mlimic	824
10	mlimic	829	require_once('../php/php-amqplib/vendor/autoload.php');
11	mlimic	824	use PhpAmqpLib\Connection\AMQPConnection;
12
13
14	mlimic	827	// Si le AMQP_DEBUG est defini a true, les messages DEBUG sont envoyes aux
15			// stdout
16			if (DEBUG > PRODUCTION_MODE) {
17			define('AMQP_DEBUG', true);
18			}
19
20
21	mlimic	824	define('HTTP_RESULT_OK', '200');
22
23
24			function parse($return)
25			{
26			// remplace les '{' et '}' par chaine vide
27			$return = str_replace('{', '', $return);
28			$return = str_replace('}', '', $return);
29			$return = trim($return, " \t");
30			// remplace les retours de ligne
31			$return = preg_replace('/\s\s+/', '', $return);
32			$return = str_replace("\"", "", $return);
33			$return = str_replace(": ", ":", $return);
34
35			$result = array();
36			$resparray = explode(',', $return);
37
38			if ($resparray) {
39			foreach ($resparray as $resp) {
40			$keyvalue = explode(':', $resp);
41			$result[trim(str_replace('"', '',$keyvalue[0]))] =
42			trim(str_replace('"', '', $keyvalue[1]));
43			}
44			}
45			return $result;
46			}
47
48			function postViaCurl($msg) {
49			$data = $msg->body;
50
51	mlimic	830	global $ERP_URL_MESSAGES;
52			if (empty($ERP_URL_MESSAGES)) {
53			logger::instance()->cleanLog();
54	mlimic	832	logger::instance()->log("Le URL de destination manque", EXTRA_VERBOSE_MODE);
55	mlimic	830	logger::instance()->writeLogToFile();
56			return;
57			}
58
59	mlimic	831	$curl_resource = curl_init($ERP_URL_MESSAGES);
60			curl_setopt($curl_resource, CURLOPT_CUSTOMREQUEST, "POST");
61			curl_setopt($curl_resource, CURLOPT_POSTFIELDS, $msg->body);
62			curl_setopt($curl_resource, CURLOPT_RETURNTRANSFER, true);
63			curl_setopt($curl_resource, CURLOPT_HTTPHEADER, array(
64	mlimic	824	'Content-Type: application/json',
65			'Content-Length: ' . strlen($msg->body))
66			);
67
68	mlimic	831	$result = curl_exec($curl_resource);
69			curl_close($curl_resource);
70	mlimic	824
71	mlimic	831
72	mlimic	824	// le resultat recu par curl est une chaine des caracteres qui contient
73			// un tableau
74			$result = parse($result);
75
76			// log le retour
77	mlimic	829	logger::instance()->cleanLog();
78	mlimic	824	logger::instance()->log($result['http_code_message'] . ': ' .
79	mlimic	960	$result['message'].' Message: '.
80			$msg->body, EXTRA_VERBOSE_MODE);
81	mlimic	824	logger::instance()->writeLogToFile();
82
83			if ($result['http_code'] == HTTP_RESULT_OK) {
84			// accuse de reception du message
85			$msg->delivery_info['channel']->
86	mlimic	952	basic_ack($msg->delivery_info['delivery_tag']);
87			} else {
88			sleep(2);
89			// accuse de reception du message
90			$msg->delivery_info['channel']->basic_reject(
91			$msg->delivery_info['delivery_tag'], true);
92	mlimic	824
93			}
94	mlimic	952	// Envoie un message avec la chaine "quit" pour canceler le consumeur.
95			if ($msg->body === 'quit') {
96			$msg->delivery_info['channel']->
97			basic_cancel($msg->delivery_info['consumer_tag']);
98			}
99
100	mlimic	824	}
101
102
103
104			// pour des besoins de logging il faut setter $_SERVER["REQUEST_URI"]
105			if (!isset($_SERVER["REQUEST_URI"])) {
106			$_SERVER["REQUEST_URI"] = __FILE__;
107			}
108
109	mlimic	961	// si le $argv[1] manque on ne peut pas cree un tag unique pour le
110			// consumeur et dans ce cas-la on termine l'execution
111			$suffix = '';
112			if (!empty($argv) && count($argv) > 1) {
113			$suffix = $argv[1];
114			} else {
115			logger::instance()->log('Missing the command line argument used for tag suffix',
116			EXTRA_VERBOSE_MODE);
117			logger::instance()->writeLogToFile();
118			exit;
119			}
120	mlimic	824
121			$exchange = 'router';
122	mlimic	830	$queue = $ERP_QUEUE;
123	mlimic	961	$consumer_tag = 'consumer'.$suffix;
124	mlimic	824
125	mlimic	830	$conn = new AMQPConnection($ERP_CONNECTION_HOST, $ERP_CONNECTIOn_PORT,
126			$ERP_CONNECTION_USER, $ERP_CONNECTION_PASSWORD,
127			$ERP_CONNECTION_VHOST);
128	mlimic	824
129			$ch = $conn->channel();
130			if (is_null($ch)) {
131			// probleme avec la connexion, donc log le fait
132			logger::instance()->log('Probleme dans l\'etablisement de la connexion',
133			EXTRA_VERBOSE_MODE);
134			logger::instance()->writeLogToFile();
135			}
136
137			$ch->queue_declare($queue, false, true, false, false);
138
139			$ch->exchange_declare($exchange, 'direct', false, true, false);
140
141			$ch->queue_bind($queue, $exchange);
142
143			$ch->basic_consume($queue, $consumer_tag, false, false, false, false, 'postViaCurl');
144
145			function shutdown($ch, $conn) {
146			$ch->close();
147			$conn->close();
148			}
149			register_shutdown_function('shutdown', $ch, $conn);
150
151			// Loop as long as the channel has callbacks registered
152			while (count($ch->callbacks)) {
153			$ch->wait();
154			}
155
156
157			?>