Skip to content

Commit 8cd0fcc

Browse files
videlalvarovidelalvaro
committed
Merge pull request php-amqplib#169 from fprochazka/patch-3
Apply PSR1 & PSR2
2 parents f36f5fb + 8da535e commit 8cd0fcc

File tree

74 files changed

+4849
-2485
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

74 files changed

+4849
-2485
lines changed

‎PhpAmqpLib/Channel/AMQPChannel.php

Lines changed: 660 additions & 212 deletions
Large diffs are not rendered by default.

‎PhpAmqpLib/Channel/AbstractChannel.php

Lines changed: 118 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -6,87 +6,126 @@
66
use PhpAmqpLib\Exception\AMQPOutOfBoundsException;
77
use PhpAmqpLib\Exception\AMQPRuntimeException;
88
use PhpAmqpLib\Helper\MiscHelper;
9-
use PhpAmqpLib\Wire\AMQPReader;
10-
use PhpAmqpLib\Message\AMQPMessage;
11-
9+
use PhpAmqpLib\Helper\Protocol\MethodMap080;
10+
use PhpAmqpLib\Helper\Protocol\MethodMap091;
1211
use PhpAmqpLib\Helper\Protocol\Protocol080;
1312
use PhpAmqpLib\Helper\Protocol\Protocol091;
1413
use PhpAmqpLib\Helper\Protocol\Wait080;
1514
use PhpAmqpLib\Helper\Protocol\Wait091;
16-
use PhpAmqpLib\Helper\Protocol\MethodMap080;
17-
use PhpAmqpLib\Helper\Protocol\MethodMap091;
15+
use PhpAmqpLib\Message\AMQPMessage;
16+
use PhpAmqpLib\Wire\AMQPReader;
1817

1918
class AbstractChannel
2019
{
20+
2121
public static $PROTOCOL_CONSTANTS_CLASS;
2222

2323
protected $debug;
24+
2425
/**
2526
*
2627
* @var AbstractConnection
2728
*/
2829
protected $connection;
2930

31+
/**
32+
* @var string
33+
*/
3034
protected $protocolVersion;
3135

36+
/**
37+
* @var \PhpAmqpLib\Helper\Protocol\Protocol091|\PhpAmqpLib\Helper\Protocol\Protocol080
38+
*/
3239
protected $protocolWriter;
3340

41+
/**
42+
* @var \PhpAmqpLib\Helper\Protocol\Wait091|\PhpAmqpLib\Helper\Protocol\Wait080
43+
*/
3444
protected $waitHelper;
3545

46+
/**
47+
* @var \PhpAmqpLib\Helper\Protocol\MethodMap091|\PhpAmqpLib\Helper\Protocol\MethodMap080
48+
*/
3649
protected $methodMap;
3750

51+
/**
52+
* @var string
53+
*/
3854
protected $channel_id;
3955

56+
/**
57+
* @var \PhpAmqpLib\Wire\AMQPReader
58+
*/
4059
protected $msg_property_reader = null;
60+
61+
/**
62+
* @var \PhpAmqpLib\Wire\AMQPReader
63+
*/
4164
protected $wait_content_reader = null;
65+
66+
/**
67+
* @var \PhpAmqpLib\Wire\AMQPReader
68+
*/
4269
protected $dispatch_reader = null;
4370

71+
72+
4473
/**
4574
* @param \PhpAmqpLib\Connection\AbstractConnection $connection
46-
* @param $channel_id
75+
* @param string $channel_id
4776
*/
4877
public function __construct(AbstractConnection $connection, $channel_id)
4978
{
5079
$this->connection = $connection;
5180
$this->channel_id = $channel_id;
5281
$connection->channels[$channel_id] = $this;
53-
$this->frame_queue = array(); // Lower level queue for frames
82+
$this->frame_queue = array(); // Lower level queue for frames
5483
$this->method_queue = array(); // Higher level queue for methods
5584
$this->auto_decode = false;
5685
$this->debug = defined('AMQP_DEBUG') ? AMQP_DEBUG : false;
5786

5887
$this->msg_property_reader = new AMQPReader(null);
5988
$this->wait_content_reader = new AMQPReader(null);
60-
$this->dispatch_reader = new AMQPReader(null);
89+
$this->dispatch_reader = new AMQPReader(null);
6190

6291
$this->protocolVersion = defined('AMQP_PROTOCOL') ? AMQP_PROTOCOL : '0.9.1';
6392
switch ($this->protocolVersion) {
64-
case '0.9.1':
65-
self::$PROTOCOL_CONSTANTS_CLASS = 'PhpAmqpLib\Wire\Constants091';
66-
$c = self::$PROTOCOL_CONSTANTS_CLASS;
67-
$this->amqp_protocol_header = $c::$AMQP_PROTOCOL_HEADER;
68-
$this->protocolWriter = new Protocol091();
69-
$this->waitHelper = new Wait091();
70-
$this->methodMap = new MethodMap091();
71-
break;
72-
case '0.8':
73-
self::$PROTOCOL_CONSTANTS_CLASS = 'PhpAmqpLib\Wire\Constants080';
74-
$c = self::$PROTOCOL_CONSTANTS_CLASS;
75-
$this->amqp_protocol_header = $c::$AMQP_PROTOCOL_HEADER;
76-
$this->protocolWriter = new Protocol080();
77-
$this->waitHelper = new Wait080();
78-
$this->methodMap = new MethodMap080();
79-
break;
80-
default:
81-
throw new AMQPRuntimeException('Protocol: ' . $this->protocolVersion . ' not implemented.');
93+
case '0.9.1':
94+
self::$PROTOCOL_CONSTANTS_CLASS = 'PhpAmqpLib\Wire\Constants091';
95+
$c = self::$PROTOCOL_CONSTANTS_CLASS;
96+
$this->amqp_protocol_header = $c::$AMQP_PROTOCOL_HEADER;
97+
$this->protocolWriter = new Protocol091();
98+
$this->waitHelper = new Wait091();
99+
$this->methodMap = new MethodMap091();
100+
break;
101+
case '0.8':
102+
self::$PROTOCOL_CONSTANTS_CLASS = 'PhpAmqpLib\Wire\Constants080';
103+
$c = self::$PROTOCOL_CONSTANTS_CLASS;
104+
$this->amqp_protocol_header = $c::$AMQP_PROTOCOL_HEADER;
105+
$this->protocolWriter = new Protocol080();
106+
$this->waitHelper = new Wait080();
107+
$this->methodMap = new MethodMap080();
108+
break;
109+
default:
110+
throw new AMQPRuntimeException('Protocol: ' . $this->protocolVersion . ' not implemented.');
82111
}
83112
}
84113

114+
115+
85116
public function getChannelId()
86117
{
87118
return $this->channel_id;
88119
}
89120

121+
122+
123+
/**
124+
* @param string $method_sig
125+
* @param string $args
126+
* @param $content
127+
* @return null|string
128+
*/
90129
public function dispatch($method_sig, $args, $content)
91130
{
92131
if (!$this->methodMap->valid_method($method_sig)) {
@@ -108,10 +147,12 @@ public function dispatch($method_sig, $args, $content)
108147
return call_user_func(array($this, $amqp_method), $this->dispatch_reader, $content);
109148
}
110149

150+
151+
111152
public function next_frame($timeout = 0)
112153
{
113154
if ($this->debug) {
114-
MiscHelper::debug_msg("waiting for a new frame");
155+
MiscHelper::debug_msg("waiting for a new frame");
115156
}
116157

117158
if (!empty($this->frame_queue)) {
@@ -121,19 +162,25 @@ public function next_frame($timeout = 0)
121162
return $this->connection->wait_channel($this->channel_id, $timeout);
122163
}
123164

124-
protected function send_method_frame($method_sig, $args="")
165+
166+
167+
protected function send_method_frame($method_sig, $args = "")
125168
{
126169
$this->connection->send_channel_method_frame($this->channel_id, $method_sig, $args);
127170
}
128171

172+
173+
129174
/**
130175
* This is here for performance reasons to batch calls to fwrite from basic.publish
131176
*/
132-
protected function prepare_method_frame($method_sig, $args="", $pkt = null)
177+
protected function prepare_method_frame($method_sig, $args = "", $pkt = null)
133178
{
134179
return $this->connection->prepare_channel_method_frame($this->channel_id, $method_sig, $args, $pkt);
135180
}
136181

182+
183+
137184
public function wait_content()
138185
{
139186
$frm = $this->next_frame();
@@ -144,58 +191,65 @@ public function wait_content()
144191
throw new AMQPRuntimeException("Expecting Content header");
145192
}
146193

147-
$this->wait_content_reader->reuse(mb_substr($payload,0,12,'ASCII'));
194+
$this->wait_content_reader->reuse(mb_substr($payload, 0, 12, 'ASCII'));
148195

149196
// $payload_reader = new AMQPReader(substr($payload,0,12));
150197
$class_id = $this->wait_content_reader->read_short();
151198
$weight = $this->wait_content_reader->read_short();
152199

153200
$body_size = $this->wait_content_reader->read_longlong();
154201

202+
//hack to avoid creating new instances of AMQPReader;
203+
$this->msg_property_reader->reuse(mb_substr($payload, 12, mb_strlen($payload, 'ASCII') - 12, 'ASCII'));
204+
155205
$msg = new AMQPMessage();
156-
$this->msg_property_reader->reuse(mb_substr($payload,12,mb_strlen($payload,'ASCII')-12,'ASCII')); //hack to avoid creating new instances of AMQPReader;
157206
$msg->load_properties($this->msg_property_reader);
158207

159208
$body_parts = array();
160209
$body_received = 0;
161-
while (bccomp($body_size,$body_received) == 1) {
210+
while (bccomp($body_size, $body_received) == 1) {
162211
$frm = $this->next_frame();
163212
$frame_type = $frm[0];
164213
$payload = $frm[1];
165214

166215
if ($frame_type != 3) {
167216
$PROTOCOL_CONSTANTS_CLASS = self::$PROTOCOL_CONSTANTS_CLASS;
168217
throw new AMQPRuntimeException("Expecting Content body, received frame type $frame_type ("
169-
.$PROTOCOL_CONSTANTS_CLASS::$FRAME_TYPES[$frame_type].")");
218+
. $PROTOCOL_CONSTANTS_CLASS::$FRAME_TYPES[$frame_type] . ")");
170219
}
171220

172221
$body_parts[] = $payload;
173222
$body_received = bcadd($body_received, mb_strlen($payload, 'ASCII'));
174223
}
175224

176-
$msg->body = implode("",$body_parts);
225+
$msg->body = implode("", $body_parts);
177226

178227
if ($this->auto_decode && isset($msg->content_encoding)) {
179228
try {
180229
$msg->body = $msg->body->decode($msg->content_encoding);
181230
} catch (\Exception $e) {
182-
if ($this->debug) {
183-
MiscHelper::debug_msg("Ignoring body decoding exception: " . $e->getMessage());
184-
}
231+
if ($this->debug) {
232+
MiscHelper::debug_msg("Ignoring body decoding exception: " . $e->getMessage());
233+
}
185234
}
186235
}
187236

188237
return $msg;
189238
}
190239

240+
241+
191242
/**
192243
* Wait for some expected AMQP methods and dispatch to them.
193244
* Unexpected methods are queued up for later calls to this PHP
194245
* method.
195246
*
247+
* @param array $allowed_methods
248+
* @param bool $non_blocking
249+
* @param int $timeout
196250
* @return mixed
197251
*/
198-
public function wait($allowed_methods=null, $non_blocking = false, $timeout = 0)
252+
public function wait($allowed_methods = null, $non_blocking = false, $timeout = 0)
199253
{
200254
$PROTOCOL_CONSTANTS_CLASS = self::$PROTOCOL_CONSTANTS_CLASS;
201255

@@ -206,23 +260,21 @@ public function wait($allowed_methods=null, $non_blocking = false, $timeout = 0)
206260
}
207261

208262
//Process deferred methods
209-
foreach ($this->method_queue as $qk=>$queued_method) {
210-
if ($this->debug) {
211-
MiscHelper::debug_msg("checking queue method " . $qk);
212-
}
263+
foreach ($this->method_queue as $qk => $queued_method) {
264+
if ($this->debug) {
265+
MiscHelper::debug_msg("checking queue method " . $qk);
266+
}
213267

214268
$method_sig = $queued_method[0];
215-
if ($allowed_methods==null || in_array($method_sig, $allowed_methods)) {
269+
if ($allowed_methods == null || in_array($method_sig, $allowed_methods)) {
216270
unset($this->method_queue[$qk]);
217271

218272
if ($this->debug) {
219-
MiscHelper::debug_msg("Executing queued method: $method_sig: " .
220-
$PROTOCOL_CONSTANTS_CLASS::$GLOBAL_METHOD_NAMES[MiscHelper::methodSig($method_sig)]);
273+
MiscHelper::debug_msg("Executing queued method: $method_sig: " .
274+
$PROTOCOL_CONSTANTS_CLASS::$GLOBAL_METHOD_NAMES[MiscHelper::methodSig($method_sig)]);
221275
}
222276

223-
return $this->dispatch($queued_method[0],
224-
$queued_method[1],
225-
$queued_method[2]);
277+
return $this->dispatch($queued_method[0], $queued_method[1], $queued_method[2]);
226278
}
227279
}
228280

@@ -234,20 +286,21 @@ public function wait($allowed_methods=null, $non_blocking = false, $timeout = 0)
234286

235287
if ($frame_type != 1) {
236288
throw new AMQPRuntimeException("Expecting AMQP method, received frame type: $frame_type ("
237-
.$PROTOCOL_CONSTANTS_CLASS::$FRAME_TYPES[$frame_type].")");
289+
. $PROTOCOL_CONSTANTS_CLASS::$FRAME_TYPES[$frame_type] . ")");
238290
}
239291

240292
if (mb_strlen($payload, 'ASCII') < 4) {
241293
throw new AMQPOutOfBoundsException("Method frame too short");
242294
}
243295

244-
$method_sig_array = unpack("n2", mb_substr($payload,0,4, 'ASCII'));
296+
$method_sig_array = unpack("n2", mb_substr($payload, 0, 4, 'ASCII'));
245297
$method_sig = "" . $method_sig_array[1] . "," . $method_sig_array[2];
246298

247-
$args = mb_substr($payload,4,mb_strlen($payload,'ASCII')-4,'ASCII');
299+
$args = mb_substr($payload, 4, mb_strlen($payload, 'ASCII') - 4, 'ASCII');
248300

249301
if ($this->debug) {
250-
MiscHelper::debug_msg("> $method_sig: " . $PROTOCOL_CONSTANTS_CLASS::$GLOBAL_METHOD_NAMES[MiscHelper::methodSig($method_sig)]);
302+
MiscHelper::debug_msg("> $method_sig: "
303+
. $PROTOCOL_CONSTANTS_CLASS::$GLOBAL_METHOD_NAMES[MiscHelper::methodSig($method_sig)]);
251304
}
252305

253306
if (in_array($method_sig, $PROTOCOL_CONSTANTS_CLASS::$CONTENT_METHODS)) {
@@ -257,15 +310,16 @@ public function wait($allowed_methods=null, $non_blocking = false, $timeout = 0)
257310
}
258311

259312
if ($allowed_methods == null ||
260-
in_array($method_sig,$allowed_methods) ||
313+
in_array($method_sig, $allowed_methods) ||
261314
in_array($method_sig, $PROTOCOL_CONSTANTS_CLASS::$CLOSE_METHODS)
262-
) {
315+
) {
263316
return $this->dispatch($method_sig, $args, $content);
264317
}
265318

266319
// Wasn't what we were looking for? save it for later
267320
if ($this->debug) {
268-
MiscHelper::debug_msg("Queueing for later: $method_sig: " . $PROTOCOL_CONSTANTS_CLASS::$GLOBAL_METHOD_NAMES[MiscHelper::methodSig($method_sig)]);
321+
MiscHelper::debug_msg("Queueing for later: $method_sig: "
322+
. $PROTOCOL_CONSTANTS_CLASS::$GLOBAL_METHOD_NAMES[MiscHelper::methodSig($method_sig)]);
269323
}
270324
$this->method_queue[] = array($method_sig, $args, $content);
271325

@@ -275,10 +329,13 @@ public function wait($allowed_methods=null, $non_blocking = false, $timeout = 0)
275329
}
276330
}
277331

278-
protected function dispatch_to_handler($handler, array $arguments)
279-
{
280-
if (is_callable($handler)) {
281-
call_user_func_array($handler, $arguments);
282-
}
283-
}
332+
333+
334+
protected function dispatch_to_handler($handler, array $arguments)
335+
{
336+
if (is_callable($handler)) {
337+
call_user_func_array($handler, $arguments);
338+
}
339+
}
340+
284341
}

‎PhpAmqpLib/Connection/AMQPConnection.php

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,5 +4,6 @@
44

55
class AMQPConnection extends AMQPStreamConnection
66
{
7+
78
// just for BC
8-
}
9+
}

0 commit comments

Comments
 (0)