modules/msgqueue/stomp_sender.inc

<?php

// $Id:$

/**
 * @file Send HTTP messages using the Stomp protocol, http://stomp.codehaus.org/Protocol
 * @author Larry Hamel, Codeguild.com
 */

/**
 * A StompMsg is a simple message object
 *
 */
class StompMsg {
    var $cmd;
    var $headers;
    var $txtmsg;

	/**
	 * constructor
	 */
    function StompMsg($cmd = NULL, $headers=NULL, $txtmsg=NULL) {
        $this->cmd = $cmd;
        $this->headers = $headers;
        $this->txtmsg = $txtmsg;
    }
}

/**
 * Stomp Connection
 *
 * Keeps connection so that command, headers and txtmsg can be transmitted for
 * any number of independent messages to the same destination.
 *
 */
class StompConn {

    var $socket;

	/**
	 * constructor connects to socket
	 */
    function StompConn($host, $port = 61613) {
        $this->socket = socket_create(AF_INET, SOCK_STREAM, 0) or die("Cannot create socket\n");
        $result = socket_connect($this->socket, $host, $port) or die("Cannot connect to server\n");
    }

	/**
	 * write a CONNECT message with credentials
	 * @return the result read from connection
	 */
    function login($userName="", $password="") {
        $this->_writeMsg( new StompMsg("CONNECT", array("login"=>$userName, "passcode"=> $password ) ) );
        return $this->readMsg();
    }

	/**
	 * send a message that should be directed to the queue indicated by the $dest param
	 * @param dest the queue or other named channel
	 * @param txtmsg the payload
	 * @param properties an optional array of headers
	 */
    function send($dest, $txtmsg, $properties=NULL) {
        $headers = array();
        if( isset($properties) ) {
            foreach ($properties as $name => $value) {
                $headers[$name] = $value;
            }
        }
        $headers["destination"] = $dest ;
        $this->_writeMsg( new StompMsg("SEND", $headers, $txtmsg) );
    }

	/**
	 * write out the given message to the socket
	 */
    function _writeMsg($stompMsg) {
        $data = $stompMsg->cmd . "\n";
        if( isset($stompMsg->headers) ) {
            foreach ($stompMsg->headers as $name => $value) {
                $data .= $name . ": " . $value . "\n";
            }
        }
        $data .= "\n";
        if( isset($stompMsg->txtmsg) ) {
            $data .= $stompMsg->txtmsg;
        }
        $l1 = strlen($data);
        $data .= "\x00\n";
        $l2 = strlen($data);

        socket_write($this->socket, $data, strlen($data)) or die("Cannot send stomp message to server\n");
    }

	/**
	 * read message from connection
	 */
    function readMsg() {
		$buffer = "";
        $in = socket_recv($this->socket, $buffer, 1, 0);

        if( $in == 0 ) {
            return NULL;
        }

        if( $in == false ) {
            return NULL;
        }

        $instring = NULL;

        while( ord($buffer) != 0  ) {

            $instring .= $buffer;
            $t = ord($buffer);

            $in = socket_recv($this->socket,$buffer,1,0);

            if( $in == 0 ) {
                return NULL;
            }

        }

        # get past the newline that follows the 0 delimitter
        $in = socket_recv($this->socket,$buffer,1,0);
        if( $in == 0 ) {
            return NULL;
        }
        if( ord($buffer) != 10 ) {
            return NULL;
        }

        list($header, $txtmsg) = explode("\n\n", $instring, 2);
        $header = explode("\n", $header);
        $allheaders = array();

        $cmd = NULL;
        foreach ($header as $oneheader) {
           if( isset($cmd) ) {
                list($name, $myval) = explode(':', $oneheader, 2);
                $allheaders[$name]=$myval;
           } else {
                $cmd = $oneheader;
           }
        }

        return new StompMsg($cmd, $allheaders, $txtmsg);
    }
}


1	<?php
2
3	// $Id:$
4
5	/**
6	* @file Send HTTP messages using the Stomp protocol, http://stomp.codehaus.org/Protocol
7	* @author Larry Hamel, Codeguild.com
8	*/
9
10	/**
11	* A StompMsg is a simple message object
12	*
13	*/
14	class StompMsg {
15	var $cmd;
16	var $headers;
17	var $txtmsg;
18
19	/**
20	* constructor
21	*/
22	function StompMsg($cmd = NULL, $headers=NULL, $txtmsg=NULL) {
23	$this->cmd = $cmd;
24	$this->headers = $headers;
25	$this->txtmsg = $txtmsg;
26	}
27	}
28
29	/**
30	* Stomp Connection
31	*
32	* Keeps connection so that command, headers and txtmsg can be transmitted for
33	* any number of independent messages to the same destination.
34	*
35	*/
36	class StompConn {
37
38	var $socket;
39
40	/**
41	* constructor connects to socket
42	*/
43	function StompConn($host, $port = 61613) {
44	$this->socket = socket_create(AF_INET, SOCK_STREAM, 0) or die("Cannot create socket\n");
45	$result = socket_connect($this->socket, $host, $port) or die("Cannot connect to server\n");
46	}
47
48	/**
49	* write a CONNECT message with credentials
50	* @return the result read from connection
51	*/
52	function login($userName="", $password="") {
53	$this->_writeMsg( new StompMsg("CONNECT", array("login"=>$userName, "passcode"=> $password ) ) );
54	return $this->readMsg();
55	}
56
57	/**
58	* send a message that should be directed to the queue indicated by the $dest param
59	* @param dest the queue or other named channel
60	* @param txtmsg the payload
61	* @param properties an optional array of headers
62	*/
63	function send($dest, $txtmsg, $properties=NULL) {
64	$headers = array();
65	if( isset($properties) ) {
66	foreach ($properties as $name => $value) {
67	$headers[$name] = $value;
68	}
69	}
70	$headers["destination"] = $dest ;
71	$this->_writeMsg( new StompMsg("SEND", $headers, $txtmsg) );
72	}
73
74	/**
75	* write out the given message to the socket
76	*/
77	function _writeMsg($stompMsg) {
78	$data = $stompMsg->cmd . "\n";
79	if( isset($stompMsg->headers) ) {
80	foreach ($stompMsg->headers as $name => $value) {
81	$data .= $name . ": " . $value . "\n";
82	}
83	}
84	$data .= "\n";
85	if( isset($stompMsg->txtmsg) ) {
86	$data .= $stompMsg->txtmsg;
87	}
88	$l1 = strlen($data);
89	$data .= "\x00\n";
90	$l2 = strlen($data);
91
92	socket_write($this->socket, $data, strlen($data)) or die("Cannot send stomp message to server\n");
93	}
94
95	/**
96	* read message from connection
97	*/
98	function readMsg() {
99	$buffer = "";
100	$in = socket_recv($this->socket, $buffer, 1, 0);
101
102	if( $in == 0 ) {
103	return NULL;
104	}
105
106	if( $in == false ) {
107	return NULL;
108	}
109
110	$instring = NULL;
111
112	while( ord($buffer) != 0 ) {
113
114	$instring .= $buffer;
115	$t = ord($buffer);
116
117	$in = socket_recv($this->socket,$buffer,1,0);
118
119	if( $in == 0 ) {
120	return NULL;
121	}
122
123	}
124
125	# get past the newline that follows the 0 delimitter
126	$in = socket_recv($this->socket,$buffer,1,0);
127	if( $in == 0 ) {
128	return NULL;
129	}
130	if( ord($buffer) != 10 ) {
131	return NULL;
132	}
133
134	list($header, $txtmsg) = explode("\n\n", $instring, 2);
135	$header = explode("\n", $header);
136	$allheaders = array();
137
138	$cmd = NULL;
139	foreach ($header as $oneheader) {
140	if( isset($cmd) ) {
141	list($name, $myval) = explode(':', $oneheader, 2);
142	$allheaders[$name]=$myval;
143	} else {
144	$cmd = $oneheader;
145	}
146	}
147
148	return new StompMsg($cmd, $allheaders, $txtmsg);
149	}
150	}
151
152