src/AsyncClient.php
public function channel(string $channel): Observable
    {
        // Only join a channel once
        if (isset($this->channels[$channel])) {
            return $this->channels[$channel];
        }

        // Ensure we only get messages for the given channel
        $channelMessages = $this->messages->filter(function (Event $event) use ($channel) {
            return $event->getChannel() !== '' && $event->getChannel() === $channel;
        });

        $subscribe = $this->connected
            ->do(function () use ($channel) {
                // Subscribe to pusher channel after connected
                $this->send(Event::subscribeOn($channel));
            })
            ->flatMapTo(Observable::empty());

        // Observable representing channel events
        $this->channels[$channel] = $channelMessages
            ->merge($subscribe)
            ->filter([Event::class, 'subscriptionSucceeded'])
            ->retryWhen(function (Observable $errors) {
                return $errors->flatMap(function (Throwable $throwable) {
                    return $this->handleLowLevelError($throwable);
                });
            })
            ->finally(function () use ($channel) {
                // Send unsubscribe event
                $this->send(Event::unsubscribeOn($channel));

                // Remove our channel from the channel list so we don't resubscribe in case we reconnect
                unset($this->channels[$channel]);
            })
            ->singleInstance();

        return $this->channels[$channel];
    }
public function channel(string $channel): Observable
    {
        // Only join a channel once
        if (isset($this->channels[$channel])) {
            return $this->channels[$channel];
        }

        // Ensure we only get messages for the given channel
        $channelMessages = $this->messages->filter(function (Event $event) use ($channel) {
            return $event->getChannel() !== '' && $event->getChannel() === $channel;
        });

        $subscribe = $this->connected
            ->do(function () use ($channel) {
                // Subscribe to pusher channel after connected
                $this->send(Event::subscribeOn($channel));
            })
            ->flatMapTo(Observable::empty());

        // Observable representing channel events
        $this->channels[$channel] = $channelMessages
            ->merge($subscribe)
            ->filter([Event::class, 'subscriptionSucceeded'])
            ->retryWhen(function (Observable $errors) {
                return $errors->flatMap(function (Throwable $throwable) {
                    return $this->handleLowLevelError($throwable);
                });
            })
            ->finally(function () use ($channel) {
                // Send unsubscribe event
                $this->send(Event::unsubscribeOn($channel));

                // Remove our channel from the channel list so we don't resubscribe in case we reconnect
                unset($this->channels[$channel]);
            })
            ->singleInstance();

        return $this->channels[$channel];
    }
public function channel ( string $ channel ) : Observable { // Only join a channel once if ( isset ( $ this -> channels [ $ channel ] ) ) { return $ this -> channels [ $ channel ] ; } // Ensure we only get messages for the given channel $ channelMessages = $ this -> messages -> filter ( function ( Event $ event ) use ( $ channel ) { return $ event -> getChannel ( ) !== '' && $ event -> getChannel ( ) === $ channel ; } ) ; $ subscribe = $ this -> connected -> do ( function ( ) use ( $ channel ) { // Subscribe to pusher channel after connected $ this -> send ( Event :: subscribeOn ( $ channel ) ) ; } ) -> flatMapTo ( Observable :: empty ( ) ) ; // Observable representing channel events $ this -> channels [ $ channel ] = $ channelMessages -> merge ( $ subscribe ) -> filter ( [ Event :: class , 'subscriptionSucceeded' ] ) -> retryWhen ( function ( Observable $ errors ) { return $ errors -> flatMap ( function ( Throwable $ throwable ) { return $ this -> handleLowLevelError ( $ throwable ) ; } ) ; } ) -> finally ( function ( ) use ( $ channel ) { // Send unsubscribe event $ this -> send ( Event :: unsubscribeOn ( $ channel ) ) ; // Remove our channel from the channel list so we don't resubscribe in case we reconnect unset ( $ this -> channels [ $ channel ] ) ; } ) -> singleInstance ( ) ; return $ this -> channels [ $ channel ] ; }
Listen on a channel.

@param  string                    $channel Channel to listen on
@throws \InvalidArgumentException
@return Observable
Listen on a channel .