-
Notifications
You must be signed in to change notification settings - Fork 1
/
pulsar-producer.go
150 lines (132 loc) · 3.48 KB
/
pulsar-producer.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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
package pulsardriver
import (
"context"
"errors"
"sync"
"time"
"github.com/apache/pulsar-client-go/pulsar"
"github.com/kafkaesque-io/pubsub-function/src/util"
log "github.com/sirupsen/logrus"
)
var producerCacheTTL = util.GetEnvInt("ProducerCacheTTL", 900)
// ProducerCache is the cache for Producer objects
var ProducerCache = util.NewCache(util.CacheOption{
TTL: time.Duration(producerCacheTTL) * time.Second,
CleanInterval: time.Duration(producerCacheTTL+2) * time.Second,
ExpireCallback: func(key string, value interface{}) {
if obj, ok := value.(*PulsarProducer); ok {
obj.Close()
} else {
log.Errorf("wrong PulsarProducer object type stored in Cache")
}
},
})
// GetPulsarProducer gets a Pulsar producer object
func GetPulsarProducer(pulsarURL, pulsarToken, topic string) (pulsar.Producer, error) {
key := pulsarURL + pulsarToken + topic
obj, exists := ProducerCache.Get(key)
if exists {
if driver, ok := obj.(*PulsarProducer); ok {
return driver.GetProducer()
}
}
prod := &PulsarProducer{
createdAt: time.Now(),
pulsarURL: pulsarURL,
token: pulsarToken,
topic: topic,
}
p, err := prod.GetProducer()
if err != nil {
// retry to close the client
if _, err = GetPulsarClient(pulsarURL, pulsarToken, true); err != nil {
return nil, err
}
if p, err = prod.GetProducer(); err != nil {
return nil, err
}
}
ProducerCache.Set(key, prod)
return p, nil
}
// PulsarProducer encapsulates the Pulsar Producer object
type PulsarProducer struct {
producer pulsar.Producer
pulsarURL string
token string
topic string
createdAt time.Time
lastUsed time.Time
sync.Mutex
}
// SendToPulsar sends data to a Pulsar producer.
func SendToPulsar(url, token, topic string, data []byte, async bool) error {
p, err := GetPulsarProducer(url, token, topic)
if err != nil {
log.Errorf("Failed to create Pulsar produce err: %v", err)
return errors.New("Failed to create Pulsar producer")
}
ctx := context.Background()
id, err := util.NewUUID()
if err != nil {
// this is very bad if happens
log.Warnf("NewUUID generation error %v", err)
id = string(time.Now().Unix())
}
prop := map[string]string{"PulsarBeamId": id}
//TODO: add cluster origin and maybe other properties
message := pulsar.ProducerMessage{
Payload: data,
EventTime: time.Now(),
Properties: prop,
}
if async {
p.SendAsync(ctx, &message, func(messageId pulsar.MessageID, msg *pulsar.ProducerMessage, err error) {
if err != nil {
log.Warnf("send to Pulsar err %v", err)
// TODO: push to a queue for retry
}
})
return nil
}
_, err = p.Send(ctx, &message)
return err
}
// GetProducer acquires a new pulsar producer
func (c *PulsarProducer) GetProducer() (pulsar.Producer, error) {
c.Lock()
defer c.Unlock()
if c.producer != nil {
return c.producer, nil
}
driver, err := GetPulsarClient(c.pulsarURL, c.token, false)
if err != nil {
return nil, err
}
p, err := driver.CreateProducer(pulsar.ProducerOptions{
Topic: c.topic,
})
if err != nil {
return nil, err
}
c.producer = p
return p, nil
}
// UpdateTime updates all time stamps in the object
func (c *PulsarProducer) UpdateTime() {
c.lastUsed = time.Now()
}
// Close closes the Pulsar client
func (c *PulsarProducer) Close() {
c.Lock()
defer c.Unlock()
if c.producer != nil {
c.producer.Close()
c.producer = nil
}
}
// Reconnect closes the current connection and reconnects again
func (c *PulsarProducer) Reconnect() (pulsar.Producer, error) {
c.Close()
return c.GetProducer()
}