PEIP_Pipe.php

Go to the documentation of this file.

<?php

/*
 * This file is part of the PEIP package.
 * (c) 2010 Timo Michna <timomichna/yahoo.de>
 * 
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

class PEIP_Pipe 
    extends PEIP_ABS_Reply_Producing_Message_Handler 
    implements 
        PEIP_INF_Channel,
        PEIP_INF_Subscribable_Channel,
        PEIP_INF_Connectable {
     
    protected 
        $eventDispatcher,
        $messageDispatcher,
        $name,
        $commands = array();

    protected static 
        $sharedEventDispatcher; 
        
    
    public function setName($name){
        $this->name = $name;
    }
    
    
    public function getName(){
        return $this->name;
    }

    
    public function send(PEIP_INF_Message $message, $timeout = -1){
        return $this->handle($message);
    }

    
    protected function doSend(PEIP_INF_Message $message){
        $this->doFireEvent('prePublish', array('MESSAGE'=>$message));
        $this->getMessageDispatcher()->notify($message);
        $this->doFireEvent('postPublish', array('MESSAGE'=>$message));
        return true;
    }
    
    
    protected function replyMessage($content){
        $message = $this->ensureMessage($content);
        if($this->getOutputChannel()){
            $this->getOutputChannel()->send($message);  
        }else{
            $this->doSend($message);            
        }           
    }
    
    
    protected function doReply(PEIP_INF_Message $message){
        $this->replyMessage($message);
    }

    
    public function subscribe(PEIP_INF_Handler $handler){
        $this->getMessageDispatcher()->connect($handler);
        $this->doFireEvent('subscribe', array('SUBSCRIBER'=>$handler));
    }
    
    
    public function unsubscribe(PEIP_INF_Handler $handler){
        $this->getMessageDispatcher()->disconnect($handler);
        $this->doFireEvent('unsubscribe', array('SUBSCRIBER'=>$handler));       
    }
    
    
    public function setMessageDispatcher(PEIP_INF_Dispatcher $dispatcher, $transferListeners = true){
        if(isset($this->dispatcher) && $transferListeners){
            foreach($this->dispatcher->getListeners() as $listener){
                $dispatcher->connect($listener);
                $this->dispatcher->disconnect($listener);       
            }   
        }
        $this->dispatcher = $dispatcher;
        $this->doFireEvent('setMessageDispatcher', array('DISPATCHER'=>$dispatcher));       
    }   
    
    
    public function getMessageDispatcher(){
        return isset($this->dispatcher) ? $this->dispatcher : $this->dispatcher = new PEIP_Dispatcher;
    }   
    
    
    public function connect($name, PEIP_INF_Handler $listener){
        $this->getEventDispatcher()->connect($name, $this, $listener);
        $this->doFireEvent('connect', array('EVENT'=>$name, 'LISTENER'=>$handler));     
    }   

    
    public function disconnect($name, PEIP_INF_Handler $listener){
        $this->getEventDispatcher()->disconnect($name, $this, $listener);
        $this->doFireEvent('disconnect', array('EVENT'=>$name, 'LISTENER'=>$handler));      
    }   

    
    public function hasListeners($name){
        return $this->getEventDispatcher()->hasListener($name, $this);      
    }

    
    public function getListeners($name){
        return $this->getEventDispatcher()->hasListener($name, $this);      
    }   
    
    
    public function connectCall($eventName, $callable){
        $this->connect($eventName, new PEIP_Callable_Handler($callable));   
    }

    
    public function disconnectCall($eventName, $callable){
        foreach($this->getEventDispatcher()->getListeners() as $handler){
            if($handler instanceof PEIP_Callable_Handler && $handler->getCallable() == $callable){
                $this->disconnect($eventName, $this, $handler); 
            }
        }
    }   

    
    protected function registerCommand($commandName, $callable){
        $this->commands[$commandName] = $callable;  
    }
    
    
    public function command(PEIP_INF_Message $cmdMessage){
        $this->doFireEvent('preCommand', array('MESSAGE'=>$cmdMessage));
        $cmd = trim((string)$cmdMessage->getHeader('COMMAND'));
        if($cmd != '' && array_key_exists($cmd, $this->commands)){
            call_user_func($this->commands[$commandName], $cmdMessage->getContent());   
        }
        $this->doFireEvent('postCommand', array('MESSAGE'=>$cmdMessage));
    }
    
    
    public function setEventDispatcher(PEIP_Object_Event_Dispatcher $dispatcher, $transferListners = true){
        if($transferListners){
            foreach($this->eventDispatcher->getEventNames($this) as $name){
                if($this->eventDispatcher->hasListeners($name, $this)){
                    foreach($this->eventDispatcher->getListeners($name, $this) as $listener){
                        $dispatcher->connect($name, $this, $listener);  
                    }
                }       
            }   
        }   
        $this->eventDispatcher = $dispatcher;   
        $this->doFireEvent('setEventDispatcher', array('DISPATCHER'=>$dispatcher)); 
    }
    
    
    public function getEventDispatcher(){
        return $this->eventDispatcher ? $this->eventDispatcher : $this->eventDispatcher = self::getSharedEventDispatcher();
    }   
    
    protected static function getSharedEventDispatcher(){
        return self::$sharedEventDispatcher ? self::$sharedEventDispatcher : self::$sharedEventDispatcher = new PEIP_Object_Event_Dispatcher; 
    }

    
    protected function doFireEvent($name, array $headers = array(), $eventClass = false){
        return $this->getEventDispatcher()->buildAndNotify($name, $this, $headers, $eventClass);
    }       
    
}   

---