I need to implement a wait timer INSIDE of a PHP React websocket event loop (perhaps multithreading?)

Question

I have a websocket application that I am building a game on, built on Ratchet which uses the React event loop. At the start of this script, I have already figured out how to implement a periodictimer, to send a pulse to the game every second, and then execute ticks and combat rounds. This works great.

However, I have recently realized that I will also need to add the ability to "lag" clients, or pause execution in a function. For example, if a player is stunned, or I want an NPC to wait for 1.5 seconds before replying to a trigger for a more "realistic" conversational feel.

Is this functionality built into the react library, or is it something that I am going to have to achieve through other means? After some research, it looks like maybe pthreads is what I may be looking for, see this question/answer: How can one use multi threading in PHP applications

To be more clear with what I am trying to achieve, take this code as an example:

    function onSay($string)
{
    global $world;

    $trigger_words = array(
        'hi',
        'hello',
        'greetings'
    );

    $triggered = false;

    foreach($trigger_words as $trigger_word)
    {
        if(stristr($string, $trigger_word))
        {
            $triggered = true;
        }
    }

    if($triggered)
    {
        foreach($world->players as $player)
        {
            if($player->pData->in_room === $this->mobile->in_room)
            {
                sleep(1);
                $this->toChar($player, $this->mobile->short . " says '`kOh, hello!``'");
            }
        }
    }
}

Obviously, this doesn't work, as the sleep(1) function will halt the entire server process.

Any insight would be greatly appreciated. Thank you!

Update: My server script:

require 'vendor/autoload.php';
require 'src/autoload.php';
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
use Ratchet\Server\IoServer;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
use React\Socket\Server as Reactor;
use React\EventLoop\Factory as LoopFactory;;

$world = new WorldInterface();

class Server implements MessageComponentInterface
{   
    public function __construct(React\EventLoop\LoopInterface $loop) 
    {
        $update = new Update();
        $update->doTick();

        $loop->addPeriodicTimer(1, function() 
        {
            $this->doBeat();    
        });
    }

public function onOpen(ConnectionInterface $ch) 
{
    global $world;
    $world->connecting[$ch->resourceId] = $ch;
    $ch->CONN_STATE = "GET_NAME";
    $ch->pData = new stdClass();
    $ch->send("Who dares storm our wayward path? ");
}

public function onMessage(ConnectionInterface $ch, $args) 
{   
    if($ch->CONN_STATE == "CONNECTED")
    {
        $ch->send("> " . $args . "\n");
        $interpreter = new Interpreter($ch);
        $interpreter->interpret($args);
    }
    else
    {
        $ch->send($args);
        $login = new Login($ch, $args);
        $login->start();
    }

}

public function onClose(ConnectionInterface $ch) 
{
    global $world;

    if(isset($ch->pData->name))
    {
        if(isset($world->players[$ch->pData->name]))
        {
            echo "Player {$ch->pData->name} has disconnected\n";
            unset($world->players->{$ch->pData->name});
        }
    }

    if(isset($world->connecting->{$ch->resourceId}))
    {
        echo "Connection " . $ch->resourceId . " has disconnected.";
        unset($world->connecting->{$ch->resourceId});
    }
}

public function onError(ConnectionInterface $conn, \Exception $e) 
{
    echo "An error has occurred: {$e->getMessage()}\n";
    $conn->close();
}

public function doBeat()
{
    global $world;
    ++$world->beats;

    foreach($world->process_queue as $trigger_beat => $process_array)
    {
        // if the beat # that the function should fire on is less than,
        // or equal to the current beat, fire the function.
        if($trigger_beat <= $world->beats)
        {
            foreach($process_array as $process)
            {
                $class = new $process->class();
                call_user_func_array(array($class, $process->function), $process->params);
            }

            // remove it from the queue
            unset($world->process_queue[$trigger_beat]);
        }
        // else, the beat # the function should fire on is greater than the current beat, 
        // so break out of the loop.
        else
        {
            break;
        }
    }

    if($world->beats % 2 === 0)
    {
        $update = new Update();
        $update->doBeat();
    }
}
}

$loop = LoopFactory::create();
$socket = new Reactor($loop);
$socket->listen(9000, 'localhost');
$server = new IoServer(new HttpServer(new WsServer(new Server($loop))),   $socket, $loop);
$server->run();

Answer 1

Alright, so I'm going to assume that because this is still unanswered there is no "easy" solution baked into the react event loop, though I would love to be wrong about that. Until then, I figured I would post my solution.

Note: I have no idea what the implications of doing this are. I have no idea how scalable it is. It is untested in a live environment with multiple processes and players.

I think it's a decent solution however. My particular game is geared toward a playerbase of maybe 20 - 30, so I think the only problem I might face is if a bunch of queued actions fire on the exact same second.

To the code!

The first thing I did (a while ago) was add a periodic timer on server startup:

public function __construct(React\EventLoop\LoopInterface $loop) 
{
    $update = new Update();
    $update->doTick();

    $loop->addPeriodicTimer(1, function() 
    {
        $this->doBeat();    
    });
}

I also have some global variables on my 'world' class:

// things in the world
public $beats = 0;
public $next_tick = 45;
public $connecting = array();
public $players = array();
public $mobiles = array();
public $objects = array();
public $mobs_in_rooms = array();
public $mobs_in_areas = array();
public $in_combat = array(
    'mobiles' => array(),
    'players' => array()
);
public $process_queue;

Note beats and process_queue.

My doBeat() function looks like this:

public function doBeat()
{
    global $world;
    ++$world->beats;

    foreach($world->process_queue as $trigger_beat => $process_array)
    {
        // if the beat # that the function should fire on is less than,
        // or equal to the current beat, fire the function.
        if($trigger_beat <= $world->beats)
        {
            foreach($process_array as $process)
            {
                $class = new $process->class();
                call_user_func_array(array($class, $process->function), $process->params);
            }

            // remove it from the queue
            unset($world->process_queue[$trigger_beat]);
        }
        // else, the beat # the function should fire on is greater than the current beat, 
        // so break out of the loop.
        else
        {
            break;
        }
    }

    print_r(array_keys($world->process_queue));

    if($world->beats % 2 === 0)
    {
        $update = new Update();
        $update->doBeat();
    }
}

Now, on my global "World" object, I have a couple other functions:

function addToProcessQueue($process_obj)
{
    //adds the process object to an array of the beat #
    //when it should be triggered on process_queue.

    $this->process_queue[(int)$process_obj->trigger_beat][] = $process_obj;
    ksort($this->process_queue);
}

function createProcessObject($array)
{
    $process_obj = new stdClass();

    if(isset($array['function']))
    {
        $process_obj->function = $array['function'];
    }
    else
    {
        echo "All process requests must define a function to call defined as a key named 'function' on the array you pass.";
    }

    if(isset($array['class']))
    {
        $process_obj->class = $array['class'];
    }
    else
    {
        echo "All process requests must define a class to call defined as a key named 'class' on the array you pass.";
    }

    if(isset($array['params']))
    {
        $process_obj->params = $array['params'];
    }
    else
    {
        $process_obj->params = array();
    }

    if(isset($array['char']))
    {
        $process_obj->char = $array['char'];
    }
    else
    {
        $process_obj->char = false;
    }

    if(isset($array['trigger_beat']) && is_numeric($array['trigger_beat']))
    {
        $process_obj->trigger_beat = $array['trigger_beat'];
    }
    else
    {
        echo "All process requests must define a trigger_beat. \n"
        . "Use world->beats to get current beat and add your wait time onto it. \n"
                . "Trigger beat MUST be an integer. \n";
    }

    $this->addToProcessQueue($process_obj);
}

Now to add a process to the queue, here is my new mobile "onSay()" command:

function onSay($string)
{
    global $world;

    $trigger_words = array(
        'hi',
        'hello',
        'greetings'
    );

    $triggered = false;

    foreach($trigger_words as $trigger_word)
    {
        if(stristr($string, $trigger_word))
        {
            $triggered = true;
        }
    }

    if($triggered)
    {
        $process_array = array(
            'trigger_beat' => $world->beats + 2,
            'function' => 'toRoom',
            'class' => 'PlayerInterface',
            'params' => array($this->mobile->in_room, $this->mobile->short . " says '`kOh, hello!``'")
        );

        $world->createProcessObject($process_array);
    }
}

So, if the mobile hears "hi", "hello" or "greetings", the "toRoom" function (which sends a string to every character in the same room) will be added to the process queue and will fire 2 seconds from when the original function was executed.

I hope all that makes sense and if anyone knows of a better way to accomplish stuff like this in php and inside an event loop please answer / comment. I'm not marking this as "correct" as like I said above, I have no idea how efficient it will be in production.

Answer 2

You can just use addTimer like you did with addPeriodicTimer. If you want to work with promises, you can create a helper promise that resolves just after your pause time.

Amp (another event loop implementation) has Amp\Pause which does exactly that. Maybe you can use that as inspiration if you want to implement a promise as mentioned.