src/EventLoop.php
public function run(string $startup_event, $startup_data = null)
    {
        $event = new Event($startup_event, $startup_data);

        $ctx = new EventContext($event, $this->event_queue);

        $loop_exit = $this->loop_exit ?? function (int $counter){
                    return $counter > $this->max_loop_count;
                };

        $loop_counter = 0;
        while($event && !($loop_exit)($loop_counter, $this->event_queue))
        {
            foreach($this->workers as $worker)
            {
                if ($worker instanceof WorkerInterface){
                    // dispatch event to worker
                    if (self::isProcessEventGenerator($worker)){
                        foreach($worker->processEvent($ctx) as $event){
                            $ctx->setEvent($event);
                            $this->dispatchImmediateEvent($ctx);
                        }
                    }
                    else{
                        $worker->processEvent($ctx);
                    }
                }
                else if (is_callable($worker)){
                    // callback worker routine
                    if (self::isFunctionGenerator($worker)){
                        foreach(($worker)($ctx) as $event){
                            $ctx->setEvent($event);
                            $this->dispatchImmediateEvent($ctx);
                        }
                    }
                    else{
                        ($worker)($ctx);
                    }
                }

                // remove singlecast event
                if ($event->isConsumed() && !$event->isMulticast()){
                    break;
                }
            }
            if ($event->isConsumed()) {
                $this->event_queue->dequeue($event);
                $ctx->setEvent($event);
            }
            $loop_counter ++;
        }
    }
public function run(string $startup_event, $startup_data = null)
    {
        $event = new Event($startup_event, $startup_data);

        $ctx = new EventContext($event, $this->event_queue);

        $loop_exit = $this->loop_exit ?? function (int $counter){
                    return $counter > $this->max_loop_count;
                };

        $loop_counter = 0;
        while($event && !($loop_exit)($loop_counter, $this->event_queue))
        {
            foreach($this->workers as $worker)
            {
                if ($worker instanceof WorkerInterface){
                    // dispatch event to worker
                    if (self::isProcessEventGenerator($worker)){
                        foreach($worker->processEvent($ctx) as $event){
                            $ctx->setEvent($event);
                            $this->dispatchImmediateEvent($ctx);
                        }
                    }
                    else{
                        $worker->processEvent($ctx);
                    }
                }
                else if (is_callable($worker)){
                    // callback worker routine
                    if (self::isFunctionGenerator($worker)){
                        foreach(($worker)($ctx) as $event){
                            $ctx->setEvent($event);
                            $this->dispatchImmediateEvent($ctx);
                        }
                    }
                    else{
                        ($worker)($ctx);
                    }
                }

                // remove singlecast event
                if ($event->isConsumed() && !$event->isMulticast()){
                    break;
                }
            }
            if ($event->isConsumed()) {
                $this->event_queue->dequeue($event);
                $ctx->setEvent($event);
            }
            $loop_counter ++;
        }
    }
public function run ( string $ startup_event , $ startup_data = null ) { $ event = new Event ( $ startup_event , $ startup_data ) ; $ ctx = new EventContext ( $ event , $ this -> event_queue ) ; $ loop_exit = $ this -> loop_exit ?? function ( int $ counter ) { return $ counter > $ this -> max_loop_count ; } ; $ loop_counter = 0 ; while ( $ event && ! ( $ loop_exit ) ( $ loop_counter , $ this -> event_queue ) ) { foreach ( $ this -> workers as $ worker ) { if ( $ worker instanceof WorkerInterface ) { // dispatch event to worker if ( self :: isProcessEventGenerator ( $ worker ) ) { foreach ( $ worker -> processEvent ( $ ctx ) as $ event ) { $ ctx -> setEvent ( $ event ) ; $ this -> dispatchImmediateEvent ( $ ctx ) ; } } else { $ worker -> processEvent ( $ ctx ) ; } } else if ( is_callable ( $ worker ) ) { // callback worker routine if ( self :: isFunctionGenerator ( $ worker ) ) { foreach ( ( $ worker ) ( $ ctx ) as $ event ) { $ ctx -> setEvent ( $ event ) ; $ this -> dispatchImmediateEvent ( $ ctx ) ; } } else { ( $ worker ) ( $ ctx ) ; } } // remove singlecast event if ( $ event -> isConsumed ( ) && ! $ event -> isMulticast ( ) ) { break ; } } if ( $ event -> isConsumed ( ) ) { $ this -> event_queue -> dequeue ( $ event ) ; $ ctx -> setEvent ( $ event ) ; } $ loop_counter ++ ; } }
Process event loop

@param string $startup_event
@param mixed $startup_data
Process event loop