Method newMessage

pubsub.go:422–485

(ctx context.Context, cn *pool.Conn, reply interface{})

}

func (c *PubSub) newMessage(ctx context.Context, cn *pool.Conn, reply interface{}) (interface{}, error) {
	switch reply := reply.(type) {
	case string:
		return &Pong{
			Payload: reply,
		}, nil
	case []interface{}:
		switch kind := reply[0].(string); kind {
		case "subscribe", "unsubscribe", "psubscribe", "punsubscribe", "ssubscribe", "sunsubscribe":
			// Can be nil in case of "unsubscribe".
			channel, _ := reply[1].(string)
			return &Subscription{
				Kind:    kind,
				Channel: channel,
				Count:   int(reply[2].(int64)),
			}, nil
		case "message", "smessage":
			channel := reply[1].(string)
			sharded := kind == "smessage"
			switch payload := reply[2].(type) {
			case string:
				msg := &Message{
					Channel: channel,
					Payload: payload,
				}
				// Record PubSub message received
				otel.RecordPubSubMessage(ctx, cn, "received", channel, sharded)
				return msg, nil
			case []interface{}:
				ss := make([]string, len(payload))
				for i, s := range payload {
					ss[i] = s.(string)
				}
				msg := &Message{
					Channel:      channel,
					PayloadSlice: ss,
				}
				// Record PubSub message received
				otel.RecordPubSubMessage(ctx, cn, "received", channel, sharded)
				return msg, nil
			default:
				return nil, fmt.Errorf("redis: unsupported pubsub message payload: %T", payload)
			}
		case "pmessage":
			channel := reply[2].(string)
			msg := &Message{
				Pattern: reply[1].(string),
				Channel: channel,
				Payload: reply[3].(string),
			}
			// Record PubSub message received (pattern message, not sharded)
			otel.RecordPubSubMessage(ctx, cn, "received", channel, false)
			return msg, nil
		case "pong":
			return &Pong{
				Payload: reply[1].(string),
			}, nil
		default: