| commit | 0e76dc2f6e8f4857d377ac9422e2df5f715b8ae2 | [log] [tgz] |
|---|---|---|
| author | wuYin <wuyinpost@gmail.com> | Tue Feb 09 10:41:58 2021 +0800 |
| committer | GitHub <noreply@github.com> | Tue Feb 09 10:41:58 2021 +0800 |
| tree | 2ebf0e8162241a8014c67d2915932c711e7a4bc4 | |
| parent | 65482772471ff028b0fbbf2306e01ec2c8b9c728 [diff] |
asynchronized send timeout checking (#460) Fixes #458 ### Motivation For current producer, send timeout checking triggered by interval batch flush If connection closed, the producer eventloop will blocked to reconnect to broker, lead to batch flush and send timeout checking take no effective, java-client timer did effective in this situation ### Modifications Asynchronized send timeout by running in independent goroutine until producer closed, and without a pending queue lock ### Verifying this change - [x] Make sure that the change passes the CI checks. ### Others Without pending queue lock, the send timeout checking gets more complicated, I don't know if it's worth it for performance.
A Go client library for the Apache Pulsar project.
This projects is developing a pure-Go client library for Pulsar that does not depend on the C++ Pulsar library.
Once feature parity and stability are reached, this will supersede the current CGo based library.
Check the Projects page at https://github.com/apache/pulsar-client-go/projects for tracking the status and the progress.
Import the client library:
import "github.com/apache/pulsar-client-go/pulsar"
Create a Producer:
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: "pulsar://localhost:6650",
})
defer client.Close()
producer, err := client.CreateProducer(pulsar.ProducerOptions{
Topic: "my-topic",
})
_, err = producer.Send(context.Background(), &pulsar.ProducerMessage{
Payload: []byte("hello"),
})
defer producer.Close()
if err != nil {
fmt.Println("Failed to publish message", err)
}
fmt.Println("Published message")
Create a Consumer:
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: "pulsar://localhost:6650",
})
defer client.Close()
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
Topic: "my-topic",
SubscriptionName: "my-sub",
Type: pulsar.Shared,
})
defer consumer.Close()
msg, err := consumer.Receive(context.Background())
if err != nil {
log.Fatal(err)
}
fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
msg.ID(), string(msg.Payload()))
Create a Reader:
client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
if err != nil {
log.Fatal(err)
}
defer client.Close()
reader, err := client.CreateReader(pulsar.ReaderOptions{
Topic: "topic-1",
StartMessageID: pulsar.EarliestMessageID(),
})
if err != nil {
log.Fatal(err)
}
defer reader.Close()
for reader.HasNext() {
msg, err := reader.Next(context.Background())
if err != nil {
log.Fatal(err)
}
fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
msg.ID(), string(msg.Payload()))
}
Contributions are welcomed and greatly appreciated. See CONTRIBUTING.md for details on submitting patches and the contribution workflow.
| Name | Scope | |||
|---|---|---|---|---|
| users@pulsar.apache.org | User-related discussions | Subscribe | Unsubscribe | Archives |
| dev@pulsar.apache.org | Development-related discussions | Subscribe | Unsubscribe | Archives |
Pulsar slack channel #dev-go at https://apache-pulsar.slack.com/
You can self-register at https://apache-pulsar.herokuapp.com/
Licensed under the Apache License, Version 2.0: http://www.apache.org/licenses/LICENSE-2.0