How to use self-describing message for protbuf

Question

One of the use cases I'm working on while using protocol buffers is to deserialize the Protocol Buffers Kafka messages which I receive at the consumer end (using sarama library and Go).

The way how i'm doing currently is i defined the sample pixel.proto file as show below.

syntax = "proto3";

package saramaprotobuf;

message Pixel {
  // Session identifier stuff
  string session_id = 2;
}

i'm sending the message through sarama.Producer(by marshalling it) receiving it sarama.Consumer (unmarshalling message it by referencing with complied pixel.proto.pb). Code is as below.

import (
    "github.com/Shopify/sarama"
    "github.com/golang/protobuf/proto"
    "log"
    "os"
    "os/signal"
    "protobuftest/example"
    "syscall"
    "time"
)

func main() {
    topic := "test_topic"
    brokerList := []string{"localhost:9092"}

    producer, err := newSyncProducer(brokerList)
    if err != nil {
        log.Fatalln("Failed to start Sarama producer:", err)
    }

    go func() {
        ticker := time.NewTicker(time.Second)
        for {
            select {
            case t := <-ticker.C:
                elliot := &example.Pixel{
                    SessionId: t.String(),
                }
                pixelToSend :=  elliot
                pixelToSendBytes, err := proto.Marshal(pixelToSend)
                if err != nil {
                    log.Fatalln("Failed to marshal example:", err)
                }

                msg := &sarama.ProducerMessage{
                    Topic: topic,
                    Value: sarama.ByteEncoder(pixelToSendBytes),
                }

                producer.SendMessage(msg)
                log.Printf("Pixel sent: %s", pixelToSend)
            }
        }

    }()

    signals := make(chan os.Signal, 1)
    signal.Notify(signals, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM)

    partitionConsumer, err := newPartitionConsumer(brokerList, topic)
    if err != nil {
        log.Fatalln("Failed to create Sarama partition consumer:", err)
    }

    log.Println("Waiting for messages...")

    for {
        select {
        case msg := <-partitionConsumer.Messages():
            receivedPixel := &example.Pixel{}
            err := proto.Unmarshal(msg.Value, receivedPixel)
            if err != nil {
                log.Fatalln("Failed to unmarshal example:", err)
            }

            log.Printf("Pixel received: %s", receivedPixel)
        case <-signals:
            log.Print("Received termination signal. Exiting.")
            return
        }
    }
}

func newSyncProducer(brokerList []string) (sarama.SyncProducer, error) {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Retry.Max = 5
    config.Producer.Return.Successes = true
    // TODO configure producer

    producer, err := sarama.NewSyncProducer(brokerList, config)
    if err != nil {
        return nil, err
    }

    return producer, nil
}

func newPartitionConsumer(brokerList []string, topic string) (sarama.PartitionConsumer, error) {
    conf := sarama.NewConfig()
    // TODO configure consumer
    consumer, err := sarama.NewConsumer(brokerList, conf)
    if err != nil {
        return nil, err
    }

    partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetOldest)
    if err != nil {
        return nil, err
    }

    return partitionConsumer, err
}

In the code as you can see I have imported the .proto file and referencing it in the main function inorder to send and receive the message. The problem here is, the solution is not generic. I will receive the message of different .proto type at the consumer end.

How can I make it generic? I know there is something called as self-describing message(dynamic message) as the part of protobuf. I referred this link https://developers.google.com/protocol-buffers/docs/techniques?csw=1#self-description . But it doesn't has any explaination on how to embed this as the part of pixel.proto(example which i have used) so that at the consumer end i came directly deserialize it to required type.

Answer 1

You would define a generic container message type that would include a DescriptorSet and an Any fields.

When sending, you build an instance of that generic message type, setting the field of type Any with an instance of your Pixel message and setting the DescriptorSet field with the DescriptorSet of the Pixel type.

That would allow the receiver of such message to parse the Any contents using the DescriptorSet you are attaching. In practical terms, this is sending a piece of proto definition together with the message. So receivers wouldn't need pre-shared proto definitions or generated code.

Having said that, I'm not sure this is what you really want because if you are planning to share proto definitions or generated code with clients then I'd suggest simply using a oneof field in a container type would be much simpler to use.