forked from redpanda-data/connect
-
Notifications
You must be signed in to change notification settings - Fork 0
/
amqp.go
69 lines (63 loc) · 2.6 KB
/
amqp.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
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/tls"
)
//------------------------------------------------------------------------------
func init() {
Constructors[TypeAMQP] = TypeSpec{
constructor: fromSimpleConstructor(NewAMQP),
Description: `
DEPRECATED: This input is deprecated and scheduled for removal in Benthos V4.
Please use [` + "`amqp_0_9`" + `](/docs/components/inputs/amqp_0_9) instead.`,
Status: docs.StatusDeprecated,
FieldSpecs: docs.FieldSpecs{
docs.FieldCommon("url",
"A URL to connect to.",
"amqp://localhost:5672/",
"amqps://guest:guest@localhost:5672/",
),
docs.FieldCommon("queue", "An AMQP queue to consume from."),
docs.FieldAdvanced("queue_declare", `
Allows you to passively declare the target queue. If the queue already exists
then the declaration passively verifies that they match the target fields.`,
).WithChildren(
docs.FieldAdvanced("enabled", "Whether to enable queue declaration.").HasDefault(false),
docs.FieldAdvanced("durable", "Whether the declared queue is durable.").HasDefault(false),
),
docs.FieldAdvanced("bindings_declare",
"Allows you to passively declare bindings for the target queue.",
[]interface{}{
map[string]interface{}{
"exchange": "foo",
"key": "bar",
},
},
).Array().WithChildren(
docs.FieldString("exchange", "The exchange of the declared binding.").HasDefault(""),
docs.FieldString("key", "The key of the declared binding.").HasDefault(""),
),
docs.FieldDeprecated("max_batch_count"),
docs.FieldCommon("consumer_tag", "A consumer tag."),
docs.FieldCommon("prefetch_count", "The maximum number of pending messages to have consumed at a time."),
docs.FieldAdvanced("prefetch_size", "The maximum amount of pending messages measured in bytes to have consumed at a time."),
tls.FieldSpec(),
},
}
}
//------------------------------------------------------------------------------
// NewAMQP creates a new AMQP input type.
// TODO: V4 Remove this.
func NewAMQP(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error) {
log.Warnln("The amqp input is deprecated, please use amqp_0_9 instead.")
a, err := reader.NewAMQP(conf.AMQP, log, stats)
if err != nil {
return nil, err
}
return NewReader("amqp", a, log, stats)
}
//------------------------------------------------------------------------------