Implement concurrency limit template

README.md

@@ -63,6 +63,7 @@ If the file is not found, gowrap will look for the template [here](https://githu
 
 List of available templates:
  - [circuitbreaker](https://github.com/hexdigest/gowrap/tree/master/templates/circuitbreaker) stops executing methods of the wrapped interface after the specified number of consecutive errors and resumes execution after the specified delay
+ - [concurrencylimit](https://github.com/hexdigest/gowrap/tree/master/templates/concurrencylimit) limits amount of simultaneous calls
  - [fallback](https://github.com/hexdigest/gowrap/tree/master/templates/fallback) takes several implementations of the source interface and concurrently runs each implementation if the previous attempt didn't return the result in a specified period of time, it returns the first non-error result
  - [log](https://github.com/hexdigest/gowrap/tree/master/templates/log) instruments the source interface with logging using standard logger from the "log" package
  - [logrus](https://github.com/hexdigest/gowrap/tree/master/templates/logrus) instruments the source interface with logging using popular [sirupsen/logrus](https://github.com/sirupsen/logrus) logger

templates/concurrencylimit

@@ -0,0 +1,46 @@
+{{ $decorator := (or .Vars.DecoratorName (printf "%sWithConcurrencyLimit" .Interface.Name)) }}
+
+import (
+ "time"
+)
+
+// {{$decorator}} implements {{.Interface.Type}}
+type {{$decorator}} struct {
+ _base {{.Interface.Type}}
+ _burst chan int
+}
+
+// New{{$decorator}} instruments an implementation of the {{.Interface.Type}} with concurrency limiting
+func New{{$decorator}}(base {{.Interface.Type}}, concurrentCalls int) *{{$decorator}} {
+ d := &{{$decorator}}{
+ _base: base,
+ _burst: make(chan int, concurrentCalls),
+ }
+
+ return d
+}
+
+{{range $method := .Interface.Methods}}
+ // {{$method.Name}} implements {{$.Interface.Type}}
+ func (_d *{{$decorator}}) {{$method.Declaration}} {
+
+ {{- if (and $method.AcceptsContext $method.ReturnsError)}}
+ select {
+ case <-ctx.Done():
+ err = ctx.Err()
+ return
+ case _d._burst<-1:
+ defer func() {
+ <-_d._burst
+ }()
+ }
+ {{else}}
+ _d._burst<-1
+ defer func() {
+ <-_d._burst
+ }()
+ {{end}}
+
+ {{ $method.Pass "_d._base." }}
+ }
+{{end}}

templates_tests/interface_with_concurrency_limit.go

@@ -0,0 +1,72 @@
+package templatestests
+
+import "context"
+
+// DO NOT EDIT!
+// This code is generated with http:https://github.com/hexdigest/gowrap tool
+// using ../templates/concurrencylimit template
+
+//go:generate gowrap gen -p github.com/hexdigest/gowrap/templates_tests -i TestInterface -t ../templates/concurrencylimit -o interface_with_concurrency_limit.go
+
+// TestInterfaceWithConcurrencyLimit implements TestInterface
+type TestInterfaceWithConcurrencyLimit struct {
+ _base TestInterface
+ _burst chan int
+}
+
+// NewTestInterfaceWithConcurrencyLimit instruments an implementation of the TestInterface with concurrency limiting
+func NewTestInterfaceWithConcurrencyLimit(base TestInterface, concurrentCalls int) *TestInterfaceWithConcurrencyLimit {
+ d := &TestInterfaceWithConcurrencyLimit{
+ _base: base,
+ _burst: make(chan int, concurrentCalls),
+ }
+
+ return d
+}
+
+// Channels implements TestInterface
+func (_d *TestInterfaceWithConcurrencyLimit) Channels(chA chan bool, chB chan<- bool, chanC <-chan bool) {
+ _d._burst <- 1
+ defer func() {
+ <-_d._burst
+ }()
+
+ _d._base.Channels(chA, chB, chanC)
+ return
+}
+
+// F implements TestInterface
+func (_d *TestInterfaceWithConcurrencyLimit) F(ctx context.Context, a1 string, a2 ...string) (result1 string, result2 string, err error) {
+ select {
+ case <-ctx.Done():
+ err = ctx.Err()
+ return
+ case _d._burst <- 1:
+ defer func() {
+ <-_d._burst
+ }()
+ }
+
+ return _d._base.F(ctx, a1, a2...)
+}
+
+// NoError implements TestInterface
+func (_d *TestInterfaceWithConcurrencyLimit) NoError(s1 string) (s2 string) {
+ _d._burst <- 1
+ defer func() {
+ <-_d._burst
+ }()
+
+ return _d._base.NoError(s1)
+}
+
+// NoParamsOrResults implements TestInterface
+func (_d *TestInterfaceWithConcurrencyLimit) NoParamsOrResults() {
+ _d._burst <- 1
+ defer func() {
+ <-_d._burst
+ }()
+
+ _d._base.NoParamsOrResults()
+ return
+}

templates_tests/interface_with_concurrency_limit_test.go

@@ -0,0 +1,46 @@
+package templatestests
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestTestInterfaceWithConcurrencyLimit_F(t *testing.T) {
+ impl := &testImpl{r1: "1", r2: "2", delay: 100 * time.Millisecond}
+
+ wrapped := NewTestInterfaceWithConcurrencyLimit(impl, 3)
+
+ for i := 0; i < 10; i++ {
+ go func() {
+ r1, r2, err := wrapped.F(context.Background(), "a1")
+ assert.NoError(t, err)
+ assert.Equal(t, "1", r1)
+ assert.Equal(t, "2", r2)
+
+ }()
+ }
+
+ <-time.After(10 * time.Millisecond)
+
+ counter := atomic.LoadUint64(&impl.callCounter)
+ assert.EqualValues(t, 3, counter) // the first burst
+
+ <-time.After(100 * time.Millisecond)
+
+ counter = atomic.LoadUint64(&impl.callCounter)
+ assert.EqualValues(t, 6, counter) // the second burst
+
+ <-time.After(100 * time.Millisecond)
+
+ counter = atomic.LoadUint64(&impl.callCounter)
+ assert.EqualValues(t, 9, counter) // the third burst
+
+ <-time.After(100 * time.Millisecond)
+
+ counter = atomic.LoadUint64(&impl.callCounter)
+ assert.EqualValues(t, 10, counter) // the 10th call
+}