Commitc6705a2

committed

Add psubscribe and punsubscribe support

1 parentb8813b7 commitc6705a2Copy full SHA for c6705a2

File tree

+70

-0

lines changed

+70

-0

lines changed

Lines changed: 27 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -113,6 +113,33 @@ public function receive(Connection $connection)`
`113`	`113`	`continue;`
`114`	`114`	`}`
`115`	`115`
	`116`	`+if ($type =='psubscribe' &&count($buffer) ==6) {`
	`117`	`+$this->resultChannel->push($buffer);`
	`118`	`+$buffer =null;`
	`119`	`+continue;`
	`120`	`+ }`
	`121`	`+`
	`122`	`+if ($type =='punsubscribe' &&count($buffer) ==6) {`
	`123`	`+$this->resultChannel->push($buffer);`
	`124`	`+$buffer =null;`
	`125`	`+continue;`
	`126`	`+ }`
	`127`	`+`
	`128`	`+if ($type =='pmessage' &&count($buffer) ==9) {`
	`129`	`+$message =newMessage();`
	`130`	`+$message->pattern =$buffer[4];`
	`131`	`+$message->channel =$buffer[6];`
	`132`	`+$message->payload =$buffer[8];`
	`133`	`+$timerID = Timer::after(30 *1000,function ()use ($message) {`
	`134`	`+static::error(sprintf('Message channel (%s) is 30 seconds full, disconnected',$message->channel));`
	`135`	`+$this->interrupt();`
	`136`	`+ });`
	`137`	`+$this->messageChannel->push($message);`
	`138`	`+ Timer::clear($timerID);`
	`139`	`+$buffer =null;`
	`140`	`+continue;`
	`141`	`+ }`
	`142`	`+`
`116`	`143`	`if ($type =='pong' &&count($buffer) ==5) {`
`117`	`144`	`$this->pingChannel->push('pong');`
`118`	`145`	`$buffer =null;`

Lines changed: 5 additions & 0 deletions

Lines changed: 38 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -125,6 +125,44 @@ public function unsubscribe(string ...$channels)`
`125`	`125`	`}`
`126`	`126`	`}`
`127`	`127`
	`128`	`+/**`
	`129`	`+ * PSubscribe.`
	`130`	`+ * @throws \Swoole\Exception`
	`131`	`+ * @throws Throwable`
	`132`	`+ */`
	`133`	`+publicfunctionpsubscribe(string ...$channels)`
	`134`	`+ {`
	`135`	`+$channels =array_map(function ($channel) {`
	`136`	`+return$this->prefix .$channel;`
	`137`	`+ },$channels);`
	`138`	`+$result =$this->commandInvoker->invoke(['psubscribe', ...$channels],count($channels));`
	`139`	`+foreach ($resultas$value) {`
	`140`	`+if ($value ===false) {`
	`141`	`+$this->commandInvoker->interrupt();`
	`142`	`+thrownewSubscribeException('Psubscribe failed');`
	`143`	`+ }`
	`144`	`+ }`
	`145`	`+ }`
	`146`	`+`
	`147`	`+/**`
	`148`	`+ * PUnsubscribe.`
	`149`	`+ * @throws \Swoole\Exception`
	`150`	`+ * @throws Throwable`
	`151`	`+ */`
	`152`	`+publicfunctionpunsubscribe(string ...$channels)`
	`153`	`+ {`
	`154`	`+$channels =array_map(function ($channel) {`
	`155`	`+return$this->prefix .$channel;`
	`156`	`+ },$channels);`
	`157`	`+$result =$this->commandInvoker->invoke(['punsubscribe', ...$channels],count($channels));`
	`158`	`+foreach ($resultas$value) {`
	`159`	`+if ($value ===false) {`
	`160`	`+$this->commandInvoker->interrupt();`
	`161`	`+thrownewUnsubscribeException('Punsubscribe failed');`
	`162`	`+ }`
	`163`	`+ }`
	`164`	`+ }`
	`165`	`+`
`128`	`166`	`/**`
`129`	`167`	`* Channel`
`130`	`168`	`* @return \Swoole\Coroutine\Channel`

Comments

(0)