Redis: Pub/Sub example no longer working

[holladay-io]

The latest updates changed pub/sub functionality. The example no longer works. The following will block waiting for messages and not pick up the first message.

package main

import (
"fmt"
"github.com/go-redis/redis"
)

func main() {
    client := redis.NewClient(&redis.Options{
        Addr:     "127.0.0.1:6379",
        })
    defer client.Close()

    pubsub := client.Subscribe("mychannel1")
    defer pubsub.Close()

    err := client.Publish("mychannel1", "hello").Err()
    if err != nil {
        panic(err)
    }

    msg, err := pubsub.ReceiveMessage()
    if err != nil {
        panic(err)
    }

    fmt.Println(msg.Channel, msg.Payload)
}

The subscriber must call ReceiveMessage() before publishing the message. This will produce the expected output.

package main

import (
"fmt"
"github.com/go-redis/redis"
)

func main() {
    client := redis.NewClient(&redis.Options{
        Addr:     "127.0.0.1:6379",
        })
    defer client.Close()

    pubsub := client.Subscribe("mychannel1")
    defer pubsub.Close()

    go func() {
        msg, err := pubsub.ReceiveMessage()
        if err != nil {
            panic(err)
        }

        fmt.Println(msg.Channel, msg.Payload)
    }()
    err := client.Publish("mychannel1", "hello").Err()
    if err != nil {
        panic(err)
    }
}

If this is the desired behavior the docs should be updated to reflect the changes

[holladay-io]

Also, does this change mean that messages can be missed while acting on the current message? (i.e. if thread isn't waiting on ReceiveMessage(), messages are dropped.)

[vmihailenco]

The first example you posted works fine for me. And AFAIK this behavior of PubSub did not change recently.

I guess your example may not work if you configure Linux to drop data that is not immediately read by the client. But again, there were no changes in go-redis regarding this behavior. It always worked like this.

[holladay-io]

Hmm.. You're right. I reinstalled the library and the example works.

However I still have an issue and I narrowed it down. I am calling Ping() to make sure I can contact Redis before performing Pub/sub. This causes the program to hang at ReceiveMessage() until I send another message.

package main

import (
"fmt"
"github.com/go-redis/redis"
)

func main() {
    client := redis.NewClient(&redis.Options{
        Addr:     "127.0.0.1:6379",
        })
    defer client.Close()

// This part causes the program to hang at ReceiveMessage()
    _, err := client.Ping().Result()
    if err != nil {
        panic(err)
    }

    pubsub := client.Subscribe("mychannel1")
    defer pubsub.Close()

    if err := client.Publish("mychannel1", "hello").Err(); err != nil {
        panic(err)
    }

    msg, err := pubsub.ReceiveMessage()
    if err != nil {
        panic(err)
    }

    fmt.Println(msg.Channel, msg.Payload)
}

I know was using a version at or before https://github.com/go-redis/redis/commit/8d52a95269db9577017b2827d8ca238388ed202f since I had to update the return values.

Am I doing something wrong by calling Ping()? Thank you for your quick reply!

[vmihailenco]

Your latest example work fine for me. And calling client.Ping should not change behavior of that program.

[holladay-io]

Sorry! Didn't mean to close this issue.

I'm running MacOS 10.12 with go 1.8.1. Here's what I tried.

I set up a clean environment, the only library in $GOPATH/src is redis. I am unable to run the example. If I comment out the ping section it works.

I pulled the docker golang image and ran inside the container. This works as expected.
docker run --rm --link redis:redis -v "$PWD":/usr/src/myapp -w /usr/src/myapp golang:1.8 /bin/bash -c 'go get github.com/go-redis/redis && go run pubsub.go'

I checked out the version prior to https://github.com/go-redis/redis/commit/8d52a95269db9577017b2827d8ca238388ed202f . This version worked with the original code (line 19 becomes pubsub, _ := client.Subscribe("mychannel1")).

rm -rf github.com
mkdir -p github.com/go-redis
cd github.com/go-redis
git clone https://github.com/go-redis/redis.git
cd redis
git checkout 3b1a641e2cb6d5f8cc1701a52721d926
...
go run pubsub.go

I'm running Redis inside a container but the same container is used for all examples. When the program hangs on ReceiveMessage() it will pick up the next message I manually publish on "mychannel1."

[vmihailenco]

It can be that first Subscribe fails for some reason. You can replace

pubsub := client.Subscribe("mychannel1")

with

pubsub := client.Subscribe()
err := pubsub.Subscribe("mychannel1")
if err != nil {
    panic(err)
}

to check for the error, but other than that I don't have any ideas...

[holladay-io]

I am led to believe this is an issue with Redis. I suspect that the subscriber does not fully register with Redis before the published message comes in.

If I add a short pause (time.Sleep()) before the publish statement it always works. However, I haven't been able to consistently reproduce the issue. Sometimes the original example hangs and sometimes not.

Thank you for your help and maintaining this library!

[vmihailenco]

I see that this is reproducible in tests:

0 clients received message
subscribed to mychannel2

want:

1 clients received message
subscribed to mychannel2
received hello from mychannel2

I will check this closer tomorrow.

[vmihailenco]

Hopefully situation improved in v6.5.1. I am still running tests locally to confirm.

[vmihailenco]

Still reproducible :( Will keep digging...

[vmihailenco]

So the problem here is that

pubsub := client.Subscribe(...)
err := client.Publish(...)

does not guarantee that Subscribe is executed before Publish, because go-redis does not read reply from Redis that Subscribe succeeded. So I guess that kernel can buffer writes to different connections and deliver them in arbitrary order. So to fix your program you need to write it like this

pubsub := client.Subscribe(...)
_, err := pubsub.Receive() // wait for subscription to be created and ignore the message
err := client.Publish(...)

Such program always works correctly.

PS AFAIK other clients have exactly the same behavior...