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();

Briard

Any reason why timer doesn't work for you? like $this->loop->addTimer(1, function() use {$player} {$this->toChar($player, $this->mobile->short . " says 'kOh, hello!``'");})` – Horney

That would be perfect, except that I don't think I have access to the loop by the time I get input from the user. I have updated my question with my server startup script. Is there a way to get access to $loop in the onMessage function? – Briard

Sure you have. You don't receive a byte from clients before you open a socket, and $loop must exist at this point, as socket requires it in the constructor. The My server script part is a mess tbh. – Horney

You didn't really help, you just told me it was possible then insulted me.. – Briard

Welp! I take it back, you did actually help. I somehow didn't realize I could just store $loop as a property, but that does work. Hopefully this makes the startup script less of a "mess" ;) If you want to put that as an answer I'll be happy to mark it correct. Thank you! – Briard

Did I? Please accept my deepest apologies if so. Believe me it wasn't intentional. If I find couple of hours to write a POC I will put it as an answer. For the time being, I only can assure you that the loop exists at the time when you receive any message from a client. – Horney

Yes you were right! And no need to apologize, my server script probably does look like a mess from an outsider honestly. I am not sure why I thought I could only use the loop inside my constructor, setting it as a property allows me to access it from anywhere, which opens up a lot of doors. Thank you again!! – Briard

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.

Briard

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.

Malinowski