Subscription->unsubscribe() throws "Subscription not found" exception

[tommyseus]

I have an issue with an unknown subscription.

I believe the nats streaming server sends an response to an UNSUB request. The \Nats\Connection::handleMSG doesn't expect this response and throws a "Subscription not found" exception.

php-nats-streaming/src/NatsStreaming/Subscription.php

Lines 257 to 260 in 4b31091

	private function closeOrUnsubscribe($doClose)
	{
	$this->stanCon->natsCon()->unsubscribe($this->sid);

[byrnedo]

Hi @tommyseus,

What version of nats streaming is it?

Could you explain your use case and flow that triggers this?

[tommyseus]

I'm sorry, the exception has a different cause.

The error is triggered by the next message. There are 2 messages published. But the subscribe handles only one (wait (1)) and than unsubscribed from the subject. Between wait and unsubscribe the second message has arrieved. I have added a sleep so that the error always occurs for demonstration.

publisher.php :

<?php

require __DIR__ . '/vendor/autoload.php';

ini_set('display_errors', '1');

$options = new Nats\ConnectionOptions();
$options->setHost('da-nats-streaming');

$streamingOptions = new \NatsStreaming\ConnectionOptions();
$streamingOptions->setClientID(uniqid('test_', false));
$streamingOptions->setClusterID("test-cluster");
$streamingOptions->setNatsOptions($options);

$streamingConnection = new \NatsStreaming\Connection($streamingOptions);
$streamingConnection->connect(60);

// publish

for ($i = 1; $i <= 4; ++$i) {
    $request = $streamingConnection->publish('dummy-subject', "publish $i  - " . date('Y-m-d H:i:s'));
    $request->wait();
    var_dump($request->gotAck());
}

subscriber.php :

<?php

require __DIR__ . '/vendor/autoload.php';

ini_set('display_errors', '1');

$options = new Nats\ConnectionOptions();
$options->setHost('da-nats-streaming');

$streamingOptions = new \NatsStreaming\ConnectionOptions();
$streamingOptions->setClientID(uniqid('dummy-client-id'));
$streamingOptions->setClusterID("test-cluster");
$streamingOptions->setNatsOptions($options);

$streamingConnection = new \NatsStreaming\Connection($streamingOptions);
$streamingConnection->connect(60);

// subscribe

$subscriptionOptions = new NatsStreaming\SubscriptionOptions();
$subscriptionOptions->setAckWaitSecs(10);
$subscriptionOptions->setManualAck(true);
$subscriptionOptions->setDurableName('dummy-durable');
$subscriptionOptions->setMaxInFlight(1);

$subscribe = $streamingConnection->queueSubscribe(
    'dummy-subject',
    'abcdefghijkl',
    function ($message) {
        /* @var $message \NatsStreaming\Msg */
        echo 'received: ' . $message->getData() . PHP_EOL;
        $message->ack();
    },
    $subscriptionOptions
);
$subscribe->wait(1);

// time to receive the next message from the nats server
sleep(4);

// triggers a wait and consumes the next message. But throws a exception because the subscriber callback was removed.
$subscribe->unsubscribe();

Versions:
nats-streamin: 0.9.2 (https://hub.docker.com/_/nats-streaming/)
byrnedo/php-nats-streaming: 0.2.4
repejota/nats: 0.8.7

[byrnedo]

Ok I think I get you. You mean that the next message is already on it's way, and hasn't been read from socket. We unsub, thus removing the sub callback in nats client. While unsubbing with stan we get this in-flight message for the now destroyed local subscription first. Have I got it?

…

On 13 June 2018 at 09:18, Tommy Seus ***@***.***> wrote: I'm sorry, the exception has a different cause. The error is triggered by the next message. There are 2 messages published. But the subscribe handles only one (wait (1)) and than unsubscribed from the subject. Between wait and unsubscribe the second message has arrieved. I have added a sleep so that the error always occurs for demonstration. publisher.php : <?php require __DIR__ . '/vendor/autoload.php'; ini_set('display_errors', '1'); $options = new Nats\ConnectionOptions(); $options->setHost('da-nats-streaming'); $streamingOptions = new \NatsStreaming\ConnectionOptions(); $streamingOptions->setClientID(uniqid('test_', false)); $streamingOptions->setClusterID("test-cluster"); $streamingOptions->setNatsOptions($options); $streamingConnection = new \NatsStreaming\Connection($streamingOptions); $streamingConnection->connect(60); // publish for ($i = 1; $i <= 4; ++$i) { $request = $streamingConnection->publish('dummy-subject', "publish $i - " . date('Y-m-d H:i:s')); $request->wait(); var_dump($request->gotAck()); } subscriber.php : <?php require __DIR__ . '/vendor/autoload.php'; ini_set('display_errors', '1'); $options = new Nats\ConnectionOptions(); $options->setHost('da-nats-streaming'); $streamingOptions = new \NatsStreaming\ConnectionOptions(); $streamingOptions->setClientID(uniqid('dummy-client-id')); $streamingOptions->setClusterID("test-cluster"); $streamingOptions->setNatsOptions($options); $streamingConnection = new \NatsStreaming\Connection($streamingOptions); $streamingConnection->connect(60); // subscribe $subscriptionOptions = new NatsStreaming\SubscriptionOptions(); $subscriptionOptions->setAckWaitSecs(10); $subscriptionOptions->setManualAck(true); $subscriptionOptions->setDurableName('dummy-durable'); $subscriptionOptions->setMaxInFlight(1); $subscribe = $streamingConnection->queueSubscribe( 'dummy-subject', 'abcdefghijkl', function ($message) { /* @var $message \NatsStreaming\Msg */ echo 'received: ' . $message->getData() . PHP_EOL; $message->ack(); }, $subscriptionOptions ); $subscribe->wait(1); // time to receive the next message from the nats server sleep(4); // triggers a wait and consumes the next message. But throws a exception because the subscriber callback was removed. $subscribe->unsubscribe(); Versions: nats-streamin: 0.9.2 (https://hub.docker.com/_/nats-streaming/) byrnedo/php-nats-streaming: 0.2.4 repejota/nats: 0.8.7 — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#7 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AFRbq392dqSSgUQmr2SYcocsqrVHAg2Uks5t8Ly1gaJpZM4UkSLQ> .

[byrnedo]

@tommyseus bump, have I grasped the flow?

[byrnedo]

@tommyseus