The Pulsar Go client

The Pulsar Go client can be used to create Pulsar producers, consumers, and readers in Go (aka Golang).

API docs available as well

For standard API docs, consult the Godoc.

Installation

Requirements

Pulsar Go client library is based on the C++ client library. Followthe instructions for C++ library for installing the binariesthrough RPM, Deb or Homebrew packages.

Installing go package

Compatibility Warning

The version number of the Go client must match the version number of the Pulsar C++ client library.

You can install the pulsar library locally using go get. Note that go get doesn't support fetching a specific tag - it will always pull in master's version of the Go client. You'll need a C++ client library that matches master.

$ go get -u github.com/apache/pulsar/pulsar-client-go/pulsar

Or you can use dep for managing the dependencies.

$ dep ensure -add github.com/apache/pulsar/pulsar-client-go/pulsar@v2.4.0

Once installed locally, you can import it into your project:

import "github.com/apache/pulsar/pulsar-client-go/pulsar"

Connection URLs

To connect to Pulsar using client libraries, you need to specify a Pulsar protocol URL.

Pulsar protocol URLs are assigned to specific clusters, use the pulsar scheme and have a default port of 6650. Here's an example for localhost:

pulsar://localhost:6650

A URL for a production Pulsar cluster may look something like this:

pulsar://pulsar.us-west.example.com:6650

If you're using TLS authentication, the URL will look like something like this:

pulsar+ssl://pulsar.us-west.example.com:6651

Creating a client

In order to interact with Pulsar, you'll first need a Client object. You can create a client object using the NewClient function, passing in a ClientOptions object (more on configuration below). Here's an example:

import (
    "log"
    "runtime"
    "github.com/apache/pulsar/pulsar-client-go/pulsar"
)
func main() {
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL: "pulsar://localhost:6650",
        OperationTimeoutSeconds: 5,
        MessageListenerThreads: runtime.NumCPU(),
    })
    if err != nil {
        log.Fatalf("Could not instantiate Pulsar client: %v", err)
    }
}

The following configurable parameters are available for Pulsar clients:

Producers

Pulsar producers publish messages to Pulsar topics. You can configure Go producers using a ProducerOptions object. Here's an example:

producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic: "my-topic",
})
if err != nil {
    log.Fatalf("Could not instantiate Pulsar producer: %v", err)
}
defer producer.Close()
msg := pulsar.ProducerMessage{
    Payload: []byte("Hello, Pulsar"),
}
if err := producer.Send(msg); err != nil {
    log.Fatalf("Producer could not send message: %v", err)
}

Blocking operation

When you create a new Pulsar producer, the operation will block (waiting on a go channel) until either a producer is successfully created or an error is thrown.

Producer operations

Pulsar Go producers have the following methods available:

Here's a more involved example usage of a producer:

import (
    "context"
    "fmt"
    "log"
    "github.com/apache/pulsar/pulsar-client-go/pulsar"
)
func main() {
    // Instantiate a Pulsar client
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL: "pulsar://localhost:6650",
    })
    if err != nil { log.Fatal(err) }
    // Use the client to instantiate a producer
    producer, err := client.CreateProducer(pulsar.ProducerOptions{
        Topic: "my-topic",
    })
    if err != nil { log.Fatal(err) }
    ctx := context.Background()
    // Send 10 messages synchronously and 10 messages asynchronously
    for i := 0; i < 10; i++ {
        // Create a message
        msg := pulsar.ProducerMessage{
            Payload: []byte(fmt.Sprintf("message-%d", i)),
        }
        // Attempt to send the message
        if err := producer.Send(ctx, msg); err != nil {
            log.Fatal(err)
        }
        // Create a different message to send asynchronously
        asyncMsg := pulsar.ProducerMessage{
            Payload: []byte(fmt.Sprintf("async-message-%d", i)),
        }
        // Attempt to send the message asynchronously and handle the response
        producer.SendAsync(ctx, asyncMsg, func(msg pulsar.ProducerMessage, err error) {
            if err != nil { log.Fatal(err) }
            fmt.Printf("the %s successfully published", string(msg.Payload))
        })
    }
}

Producer configuration

Consumers

Pulsar consumers subscribe to one or more Pulsar topics and listen for incoming messages produced on that topic/those topics. You can configure Go consumers using a ConsumerOptions object. Here's a basic example that uses channels:

msgChannel := make(chan pulsar.ConsumerMessage)
consumerOpts := pulsar.ConsumerOptions{
    Topic:            "my-topic",
    SubscriptionName: "my-subscription-1",
    Type:             pulsar.Exclusive,
    MessageChannel:   msgChannel,
}
consumer, err := client.Subscribe(consumerOpts)
if err != nil {
    log.Fatalf("Could not establish subscription: %v", err)
}
defer consumer.Close()
for cm := range msgChannel {
    msg := cm.Message
    fmt.Printf("Message ID: %s", msg.ID())
    fmt.Printf("Message value: %s", string(msg.Payload()))
    consumer.Ack(msg)
}

Blocking operation

When you create a new Pulsar consumer, the operation will block (on a go channel) until either a producer is successfully created or an error is thrown.

Consumer operations

Pulsar Go consumers have the following methods available:

Receive example

Here's an example usage of a Go consumer that uses the Receive() method to process incoming messages:

import (
    "context"
    "log"
    "github.com/apache/pulsar/pulsar-client-go/pulsar"
)
func main() {
    // Instantiate a Pulsar client
    client, err := pulsar.NewClient(pulsar.ClientOptions{
            URL: "pulsar://localhost:6650",
    })
    if err != nil { log.Fatal(err) }
    // Use the client object to instantiate a consumer
    consumer, err := client.Subscribe(pulsar.ConsumerOptions{
        Topic:            "my-golang-topic",
        SubscriptionName: "sub-1",
        SubscriptionType: pulsar.Exclusive,
    })
    if err != nil { log.Fatal(err) }
    defer consumer.Close()
    ctx := context.Background()
    // Listen indefinitely on the topic
    for {
        msg, err := consumer.Receive(ctx)
        if err != nil { log.Fatal(err) }
        // Do something with the message
        err = processMessage(msg)
        if err == nil {
            // Message processed successfully
            consumer.Ack(msg)
        } else {
            // Failed to process messages
            consumer.Nack(msg)
        }
    }
}

Consumer configuration

Readers

Pulsar readers process messages from Pulsar topics. Readers are different from consumers because with readers you need to explicitly specify which message in the stream you want to begin with (consumers, on the other hand, automatically begin with the most recent unacked message). You can configure Go readers using a ReaderOptions object. Here's an example:

reader, err := client.CreateReader(pulsar.ReaderOptions{
    Topic: "my-golang-topic",
    StartMessageId: pulsar.LatestMessage,
})

Blocking operation

When you create a new Pulsar reader, the operation will block (on a go channel) until either a reader is successfully created or an error is thrown.

Reader operations

Pulsar Go readers have the following methods available:

"Next" example

Here's an example usage of a Go reader that uses the Next() method to process incoming messages:

import (
    "context"
    "log"
    "github.com/apache/pulsar/pulsar-client-go/pulsar"
)
func main() {
    // Instantiate a Pulsar client
    client, err := pulsar.NewClient(pulsar.ClientOptions{
            URL: "pulsar://localhost:6650",
    })
    if err != nil { log.Fatalf("Could not create client: %v", err) }
    // Use the client to instantiate a reader
    reader, err := client.CreateReader(pulsar.ReaderOptions{
        Topic:          "my-golang-topic",
        StartMessageID: pulsar.EarliestMessage,
    })
    if err != nil { log.Fatalf("Could not create reader: %v", err) }
    defer reader.Close()
    ctx := context.Background()
    // Listen on the topic for incoming messages
    for {
        msg, err := reader.Next(ctx)
        if err != nil { log.Fatalf("Error reading from topic: %v", err) }
        // Process the message
    }
}

In the example above, the reader begins reading from the earliest available message (specified by pulsar.EarliestMessage). The reader can also begin reading from the latest message (pulsar.LatestMessage) or some other message ID specified by bytes using the DeserializeMessageID function, which takes a byte array and returns a MessageID object. Here's an example:

lastSavedId := // Read last saved message id from external store as byte[]
reader, err := client.CreateReader(pulsar.ReaderOptions{
    Topic:          "my-golang-topic",
    StartMessageID: DeserializeMessageID(lastSavedId),
})

Reader configuration

Messages

The Pulsar Go client provides a ProducerMessage interface that you can use to construct messages to producer on Pulsar topics. Here's an example message:

msg := pulsar.ProducerMessage{
    Payload: []byte("Here is some message data"),
    Key: "message-key",
    Properties: map[string]string{
        "foo": "bar",
    },
    EventTime: time.Now(),
    ReplicationClusters: []string{"cluster1", "cluster3"},
}
if err := producer.send(msg); err != nil {
    log.Fatalf("Could not publish message due to: %v", err)
}

The following methods parameters are available for ProducerMessage objects:

TLS encryption and authentication

In order to use TLS encryption, you'll need to configure your client to do so:

• Use pulsar+ssl URL type
• Set TLSTrustCertsFilePath to the path to the TLS certs used by your client and the Pulsar broker
• Configure Authentication optionHere's an example:

opts := pulsar.ClientOptions{
    URL: "pulsar+ssl://my-cluster.com:6651",
    TLSTrustCertsFilePath: "/path/to/certs/my-cert.csr",
    Authentication: NewAuthenticationTLS("my-cert.pem", "my-key.pem"),
}