forked from redpanda-data/connect
-
Notifications
You must be signed in to change notification settings - Fork 0
/
amqp_1.go
65 lines (56 loc) · 2.01 KB
/
amqp_1.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
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/metrics"
"github.com/dafanshu/benthos/v3/lib/types"
"github.com/dafanshu/benthos/v3/lib/util/amqp/sasl"
"github.com/dafanshu/benthos/v3/lib/util/tls"
)
//------------------------------------------------------------------------------
func init() {
Constructors[TypeAMQP1] = TypeSpec{
constructor: fromSimpleConstructor(NewAMQP1),
Status: docs.StatusBeta,
Summary: `
Reads messages from an AMQP (1.0) server.`,
Description: `
### Metadata
This input adds the following metadata fields to each message:
` + "``` text" + `
- amqp_content_type
- amqp_content_encoding
- amqp_creation_time
- All string typed message annotations
` + "```" + `
You can access these metadata fields using
[function interpolation](/docs/configuration/interpolation#metadata).`,
Categories: []Category{
CategoryServices,
},
FieldSpecs: docs.FieldSpecs{
docs.FieldCommon("url",
"A URL to connect to.",
"amqp://localhost:5672/",
"amqps://guest:guest@localhost:5672/",
),
docs.FieldCommon("source_address", "The source address to consume from.", "/foo", "queue:/bar", "topic:/baz"),
docs.FieldAdvanced("azure_renew_lock", "Experimental: Azure service bus specific option to renew lock if processing takes more then configured lock time").AtVersion("3.45.0"),
tls.FieldSpec(),
sasl.FieldSpec(),
},
}
}
//------------------------------------------------------------------------------
// NewAMQP1 creates a new AMQP1 input type.
func NewAMQP1(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error) {
var a reader.Async
var err error
if a, err = reader.NewAMQP1(conf.AMQP1, log, stats); err != nil {
return nil, err
}
a = reader.NewAsyncBundleUnacks(a)
return NewAsyncReader(TypeAMQP1, true, a, log, stats)
}
//------------------------------------------------------------------------------