public function push($message)
    {
        $message = json_decode($message, true);
        $topic = array_get($this->getTopics(), $message['topic']);
        $data = $message['data'];
        $recipient = array_get($message, 'recipient');
        
        // Handle TwoStream::stop();
        if ($message['topic'] == 'cupoftea/twostream/server/stop') {
            if ($data['secret'] == config('app.key')) {
                event(new ServerStopped());
                $this->line('Stop Command fired.');
                $this->question('Stopping Server.');
                
                die();
            } else {
                // report malicious attempt
            }
        }
        
        if (! $topic) {
            return;
        }
        
        $this->setTopic($topic);
        
        if ($recipient == 'all') {
            $topic->broadcast($data);
        } else {
            foreach ((array) $recipient as $recipient) {
                if (($sessionId = $this->getUserSessionId($recipient)) || ($sessionId = WsSession::isValidId($recipient) ? $recipient : false)) {
                    foreach ($topic->getIterator() as $client) {
                        if ($client->session == $sessionId) {
                            $client->event($topic->getId(), $data);
                        }
                    }
                } else {
                    throw new InvalidRecipientException($recipient);
                }
            }
        }
    }