forked from redpanda-data/connect
-
Notifications
You must be signed in to change notification settings - Fork 0
/
nsq.go
56 lines (50 loc) · 1.9 KB
/
nsq.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
package input
import (
"github.com/dafanshu/benthos/v3/internal/docs"
"github.com/dafanshu/benthos/v3/lib/input/reader"
"github.com/dafanshu/benthos/v3/lib/log"
"github.com/dafanshu/benthos/v3/lib/message/batch"
"github.com/dafanshu/benthos/v3/lib/metrics"
"github.com/dafanshu/benthos/v3/lib/types"
"github.com/dafanshu/benthos/v3/lib/util/tls"
)
//------------------------------------------------------------------------------
func init() {
Constructors[TypeNSQ] = TypeSpec{
constructor: fromSimpleConstructor(NewNSQ),
Summary: `
Subscribe to an NSQ instance topic and channel.`,
FieldSpecs: docs.FieldSpecs{
func() docs.FieldSpec {
b := batch.FieldSpec()
b.IsDeprecated = true
return b
}(),
docs.FieldCommon("nsqd_tcp_addresses", "A list of nsqd addresses to connect to.").Array(),
docs.FieldCommon("lookupd_http_addresses", "A list of nsqlookupd addresses to connect to.").Array(),
tls.FieldSpec(),
docs.FieldCommon("topic", "The topic to consume from."),
docs.FieldCommon("channel", "The channel to consume from."),
docs.FieldCommon("user_agent", "A user agent to assume when connecting."),
docs.FieldCommon("max_in_flight", "The maximum number of pending messages to consume at any given time."),
},
Categories: []Category{
CategoryServices,
},
}
}
//------------------------------------------------------------------------------
// NewNSQ creates a new NSQ input type.
func NewNSQ(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error) {
var n reader.Async
var err error
if n, err = reader.NewNSQ(conf.NSQ, log, stats); err != nil {
return nil, err
}
if n, err = reader.NewAsyncBatcher(conf.NSQ.Batching, n, mgr, log, stats); err != nil {
return nil, err
}
n = reader.NewAsyncBundleUnacks(n)
return NewAsyncReader(TypeNSQ, true, n, log, stats)
}
//------------------------------------------------------------------------------