Ability to close a subscribe at any time (or by condition)

[roimee6]

Right now when I do that:

$relay->subscribe(["channel"], function ($relay, $channel, $message) {

});

I can't unsuscribe directly because it's an infinite loop I guess
So I propose as a likely function:

$relay->unsubscribeWithCondition(["channel"], function(): bool {
   return $this->isKilled;
});

The function name is an example
I think you can find another way to do what I want, but I think it would be a good idea to do something like this

Thanks

[tillkruss]

@michael-grunder Any thoughts?

[yatsukhnenko]

@tillkruss Let's add ability to unsubscribe inside callback like in PhpRedis

[roimee6]

@tillkruss Let's add ability to unsubscribe inside callback like in PhpRedis

Good idea 😁 Please let me know when you've added the function

[roimee6]

You say this function exists in phpredis? May I have the name, please?

[michael-grunder]

Hi @roimee6

You can unsubscribe inside the callback. Here's an example program that will do just that:

<?php

$relay = new \Relay\Relay('localhost', 6379);

$relay->subscribe(['foo'], function ($relay, $channel, $message) {
    if ($message == 'unsubscribe') {
        printf("Unsubscribing from %s\n", $channel);
        $relay->unsubscribe([$channel]);
    } else {
        printf("%s => %s\n", $channel, $message);
    }
});

printf("Finished with subscribe loop, exiting\n");

You can verify this by publishing unsubscribe to the channel in another terminal session.

This also works if you're subscribed to multiple channels:

<?php

$relay = new \Relay\Relay('localhost', 6379);

$relay->subscribe(['one', 'two', 'three'], function ($relay, $channel, $message) {
    if ($message == 'unsubscribe') {
        printf("Unsubscribing from %s\n", $channel);
        $relay->unsubscribe([$channel]);
    } else if ($message == 'unsubscribe-all') {
        printf("Unsubscribing from all *\n");
        $relay->unsubscribe([]);
    } else {
        printf("%s => %s\n", $channel, $message);
    }
});

printf("Finished with subscribe loop, exiting\n");

Relay can do other cool things such as being able to subscribe to a channel within the subscribe callback without recursing the call-stack, meaning you could in theory subscribe to a "control" channel and then subscribe and unsubscribe to other channels based on the messages.

[roimee6]

Hi @michael-grunder ! I had already thought of it and I actually did it, I still propose the idea because it may be simpler since my idea to do it

$database->subscribe(["nitro"], function ($_, $__, $message) use ($database, $notifier) {
   if ($message === strval($this->getThreadId())) {
      $database->unsubscribe(["nitro"]);
      return;
   }

   // My code
});

I don't mind doing it, but it's not the best solution, so I'm proposing a function to do it more "cleanly". And also, when I do this, I get an error that completely stops my code:

free(): double free detected in tcache 2
Aborted

Thanks for suggestion anyway

[michael-grunder]

Can you provide code that can replicate that double-free? That is a bug that we'll certainly fix. Also what version of Relay are you using?

Also I'm not opposed to providing a mechanism like you're interested in. I'll have a chat about it with @tillkruss

[roimee6]

Can you provide code that can replicate that double-free? That is a bug that we'll certainly fix. Also what version of Relay are you using?

Also I'm not opposed to providing a mechanism like you're interested in. I'll have a chat about it with @tillkruss

Double-free:

When my server stops I have a function called "onDisable" which is called, here I call a function "close" which will call the published ID of a thread started at the beginning of the server . The thread contains the subscribe and unsubscribe the channel (the code I sent above)

that's all

for the code:

Main:

protected function onDisable(): void
{
    Database::getInstance()->close();
}

Database:

public function close(): void
{
    $this->getDatabase()->publish("nitro", $this->thread->getThreadId());
    $this->getDatabase()->close();
}

My thread:

use pmmp\thread\ThreadSafeArray;
use pocketmine\snooze\SleeperHandlerEntry;
use pocketmine\thread\Thread;
use Relay\Relay;

class SubscribeThread extends Thread
{
    public function __construct(
        private ThreadSafeArray     $redis,
        protected ThreadSafeArray   $buffer,
        private SleeperHandlerEntry $sleeperEntry
    )
    {
    }

    public function onRun(): void
    {
        $redis = $this->redis;

        $database = new Relay;
        $database->connect($redis->shift());
        $database->auth($redis->shift());

        $notifier = $this->sleeperEntry->createNotifier();

        $database->subscribe(["nitro"], function ($_, $__, $message) use ($database, $notifier) {
            if ($message === strval($this->getThreadId())) {
                $database->unsubscribe(["nitro"]);
                return;
            }

            // Until code
        });
    }
}

Anything I can provide that will be useful to you
for more info about threads: github.com/pmmp

[roimee6]

And the relay version is v0.6.5

[michael-grunder]

Ahh interesting. Relay isn't thread-safe so using it this way could cause all sorts of issues. I will run it locally however and see exactly where/how this double-free is occurring. It may be fixable but then again maybe not. I will post an update here when I have one.