forked from redpanda-data/connect
-
Notifications
You must be signed in to change notification settings - Fork 0
/
throttle.go
119 lines (96 loc) · 3.8 KB
/
throttle.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
// Copyright (c) 2018 Ashley Jeffs
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package processor
import (
"fmt"
"time"
"github.com/Jeffail/benthos/lib/log"
"github.com/Jeffail/benthos/lib/metrics"
"github.com/Jeffail/benthos/lib/types"
)
//------------------------------------------------------------------------------
func init() {
Constructors[TypeThrottle] = TypeSpec{
constructor: NewThrottle,
description: `
Throttles the throughput of a pipeline to a maximum of one message batch per
period. This throttle is per processing pipeline, and therefore four threads
each with a throttle would result in four times the rate specified.
The period should be specified as a time duration string. For example, '1s'
would be 1 second, '10ms' would be 10 milliseconds, etc.`,
}
}
//------------------------------------------------------------------------------
// ThrottleConfig contains configuration fields for the Throttle processor.
type ThrottleConfig struct {
Period string `json:"period" yaml:"period"`
}
// NewThrottleConfig returns a ThrottleConfig with default values.
func NewThrottleConfig() ThrottleConfig {
return ThrottleConfig{
Period: "100us",
}
}
//------------------------------------------------------------------------------
// Throttle is a processor that limits the stream of a pipeline to one message
// batch per period specified.
type Throttle struct {
conf Config
log log.Modular
stats metrics.Type
duration time.Duration
lastBatch time.Time
mCount metrics.StatCounter
mSent metrics.StatCounter
mSentParts metrics.StatCounter
}
// NewThrottle returns a Throttle processor.
func NewThrottle(
conf Config, mgr types.Manager, log log.Modular, stats metrics.Type,
) (Type, error) {
t := &Throttle{
conf: conf,
log: log.NewModule(".processor.throttle"),
stats: stats,
mCount: stats.GetCounter("processor.throttle.count"),
mSent: stats.GetCounter("processor.throttle.sent"),
mSentParts: stats.GetCounter("processor.throttle.parts.sent"),
}
var err error
if t.duration, err = time.ParseDuration(conf.Throttle.Period); err != nil {
return nil, fmt.Errorf("failed to parse period: %v", err)
}
return t, nil
}
//------------------------------------------------------------------------------
// ProcessMessage applies the processor to a message, either creating >0
// resulting messages or a response to be sent back to the message source.
func (m *Throttle) ProcessMessage(msg types.Message) ([]types.Message, types.Response) {
m.mCount.Incr(1)
if since := time.Since(m.lastBatch); m.duration > since {
time.Sleep(m.duration - since)
}
m.lastBatch = time.Now()
m.mSent.Incr(1)
m.mSentParts.Incr(int64(msg.Len()))
msgs := [1]types.Message{msg}
return msgs[:], nil
}
//------------------------------------------------------------------------------