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'], 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__;
}


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

$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			$result['message'], EXTRA_VERBOSE_MODE);
80			logger::instance()->writeLogToFile();
81
82			if ($result['http_code'] == HTTP_RESULT_OK) {
83			// accuse de reception du message
84			$msg->delivery_info['channel']->
85	mlimic	952	basic_ack($msg->delivery_info['delivery_tag']);
86			} else {
87			sleep(2);
88			// accuse de reception du message
89			$msg->delivery_info['channel']->basic_reject(
90			$msg->delivery_info['delivery_tag'], true);
91	mlimic	824
92			}
93	mlimic	952	// Envoie un message avec la chaine "quit" pour canceler le consumeur.
94			if ($msg->body === 'quit') {
95			$msg->delivery_info['channel']->
96			basic_cancel($msg->delivery_info['consumer_tag']);
97			}
98
99	mlimic	824	}
100
101
102
103			// pour des besoins de logging il faut setter $_SERVER["REQUEST_URI"]
104			if (!isset($_SERVER["REQUEST_URI"])) {
105			$_SERVER["REQUEST_URI"] = __FILE__;
106			}
107
108
109
110			$exchange = 'router';
111	mlimic	830	$queue = $ERP_QUEUE;
112	mlimic	824	$consumer_tag = 'consumer';
113
114	mlimic	830	$conn = new AMQPConnection($ERP_CONNECTION_HOST, $ERP_CONNECTIOn_PORT,
115			$ERP_CONNECTION_USER, $ERP_CONNECTION_PASSWORD,
116			$ERP_CONNECTION_VHOST);
117	mlimic	824
118			$ch = $conn->channel();
119			if (is_null($ch)) {
120			// probleme avec la connexion, donc log le fait
121			logger::instance()->log('Probleme dans l\'etablisement de la connexion',
122			EXTRA_VERBOSE_MODE);
123			logger::instance()->writeLogToFile();
124			}
125
126			$ch->queue_declare($queue, false, true, false, false);
127
128			$ch->exchange_declare($exchange, 'direct', false, true, false);
129
130			$ch->queue_bind($queue, $exchange);
131
132			$ch->basic_consume($queue, $consumer_tag, false, false, false, false, 'postViaCurl');
133
134			function shutdown($ch, $conn) {
135			$ch->close();
136			$conn->close();
137			}
138			register_shutdown_function('shutdown', $ch, $conn);
139
140			// Loop as long as the channel has callbacks registered
141			while (count($ch->callbacks)) {
142			$ch->wait();
143			}
144
145
146			?>