-
Notifications
You must be signed in to change notification settings - Fork 0
/
pulsar.go
57 lines (51 loc) · 1.35 KB
/
pulsar.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
package gateway
import (
"context"
"github.com/apache/pulsar-client-go/pulsar"
"github.com/jponge/playground-go-microservices/pulsar/temperature-store-gateway/data"
"log"
"time"
)
func (service *Service) initPulsarClient() {
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: service.PulsarURL,
OperationTimeout: 30 * time.Second,
ConnectionTimeout: 30 * time.Second,
})
if err != nil {
log.Fatalf("Could not create the Pulsar client: %v", err)
}
service.pulsarClient = client
producer, err := client.CreateProducer(pulsar.ProducerOptions{
Topic: service.PulsarTopic,
//DisableBatching: true,
})
if err != nil {
log.Fatalf("Could not create the Pulsar producer: %v", err)
}
service.pulsarProducer = producer
log.Println("🚀 Pulsar client initialized")
}
func (service Service) closePulsarClient() {
if service.pulsarProducer != nil {
service.pulsarProducer.Close()
}
if service.pulsarClient != nil {
service.pulsarClient.Close()
}
}
func (service Service) pushToPulsar(ctx context.Context, payload data.Payload) error {
jsonBytes, err := payload.ToJSON()
if err != nil {
return err
}
id, err := service.pulsarProducer.Send(ctx, &pulsar.ProducerMessage{
Key: payload.SensorID,
Payload: jsonBytes,
})
if err != nil {
return err
}
log.Printf("Sent to Pulsar with message id: %s", id)
return nil
}