gen: sender/httpsender refactoring/cleanup

mesos · Sep 10, 2017 · a088793 · a088793
1 parent 1bedc2a
commit a088793
Show file tree

Hide file tree

Showing 7 changed files with 140 additions and 81 deletions.
diff --git a/api/v1/lib/agent/calls/calls_generated.go b/api/v1/lib/agent/calls/calls_generated.go
@@ -5,6 +5,7 @@ package calls
 
 import (
 	"context"
+	"sync/atomic"
 
 	"github.com/mesos/mesos-go/api/v1/lib"
 	"github.com/mesos/mesos-go/api/v1/lib/encoding"
@@ -43,26 +44,47 @@ type (
 )
 
 func (f RequestFunc) Call() *agent.Call { return f() }
+
 func (f RequestFunc) Marshaler() encoding.Marshaler {
-	call := f()
 	// avoid returning (*agent.Call)(nil) for interface type
-	if call != nil {
+	if call := f(); call != nil {
 		return call
 	}
 	return nil
 }
 
-func (f RequestStreamingFunc) Call() *agent.Call { return f() }
+func (f RequestStreamingFunc) Push(c ...*agent.Call) RequestStreamingFunc { return Push(f, c...) }
+
 func (f RequestStreamingFunc) Marshaler() encoding.Marshaler {
-	call := f()
 	// avoid returning (*agent.Call)(nil) for interface type
-	if call != nil {
+	if call := f(); call != nil {
 		return call
 	}
 	return nil
 }
+
 func (f RequestStreamingFunc) IsStreaming() {}
 
+func (f RequestStreamingFunc) Call() *agent.Call { return f() }
+
+// Push prepends one or more calls onto a request stream. If no calls are given then the original stream is returned.
+func Push(r RequestStreaming, c ...*agent.Call) RequestStreamingFunc {
+	if len(c) == 0 {
+		return r.Call
+	}
+	var forward int32
+	return func() *agent.Call {
+		if atomic.LoadInt32(&forward) == 1 {
+			return Push(r, c[1:]...).Call()
+		}
+		atomic.StoreInt32(&forward, 1)
+		return c[0]
+	}
+}
+
+// Empty generates a stream that always returns nil.
+func Empty() RequestStreamingFunc { return func() *agent.Call { return nil } }
+
 var (
 	_ = Request(RequestFunc(nil))
 	_ = RequestStreaming(RequestStreamingFunc(nil))
@@ -80,11 +102,10 @@ func FromChan(ch <-chan *agent.Call) RequestStreamingFunc {
 		return func() *agent.Call { return nil }
 	}
 	return func() *agent.Call {
-		m, ok := <-ch
-		if !ok {
-			return nil
+		if m, ok := <-ch; ok {
+			return m
 		}
-		return m
+		return nil
 	}
 }
 
@@ -93,12 +114,20 @@ func (f SenderFunc) Send(ctx context.Context, r Request) (mesos.Response, error)
 	return f(ctx, r)
 }
 
+// IgnoreResponse generates a sender that closes any non-nil response received by Mesos.
+func IgnoreResponse(s Sender) SenderFunc {
+	return func(ctx context.Context, r Request) (mesos.Response, error) {
+		resp, err := s.Send(ctx, r)
+		if resp != nil {
+			resp.Close()
+		}
+		return nil, err
+	}
+}
+
 // SendNoData is a convenience func that executes the given Call using the provided Sender
 // and always drops the response data.
-func SendNoData(ctx context.Context, sender Sender, r Request) error {
-	resp, err := sender.Send(ctx, r)
-	if resp != nil {
-		resp.Close()
-	}
-	return err
+func SendNoData(ctx context.Context, sender Sender, r Request) (err error) {
+	_, err = IgnoreResponse(sender).Send(ctx, r)
+	return
 }
diff --git a/api/v1/lib/agent/calls/calls_test.go b/api/v1/lib/agent/calls/calls_test.go
@@ -59,18 +59,11 @@ func Example() {
 		AttachContainerOutput(mesos.ContainerID{}),
 	)
 
-	blackhole = func(calls ...*agent.Call) {
-		ch := make(chan *agent.Call, len(calls))
-		for i := range calls {
-			ch <- calls[i]
-		}
-		swallow(sender.Send(ctx, FromChan(ch)))
-	}
-	blackhole(
+	swallow(sender.Send(ctx, Empty().Push(
 		AttachContainerInput(mesos.ContainerID{}),
 		AttachContainerInputTTY(nil),
 		AttachContainerInputData(nil),
-	)
+	)))
 
 	// Output:
 }
diff --git a/api/v1/lib/extras/gen/httpsender.go b/api/v1/lib/extras/gen/httpsender.go
@@ -47,16 +47,9 @@ func NewSender(c *httpcli.Client) calls.Sender {
 
 		var req client.Request
 
-		switch r.(type) {
+		switch r := r.(type) {
 		case calls.RequestStreaming:
-			first := true
-			req = calls.RequestStreamingFunc(func() {{.Type "C"}} {
-				if first {
-					first = false
-					return obj
-				}
-				return r.Call()
-			})
+			req = calls.Push(r, obj)
 		default:
 			req = calls.NonStreaming(obj)
 		}

diff --git a/api/v1/lib/extras/gen/sender.go b/api/v1/lib/extras/gen/sender.go
@@ -18,6 +18,7 @@ var handlersTemplate = template.Must(template.New("").Parse(`package {{.Package}
 
 import (
 	"context"
+	"sync/atomic"
 
 	"github.com/mesos/mesos-go/api/v1/lib"
 	"github.com/mesos/mesos-go/api/v1/lib/encoding"
@@ -58,26 +59,47 @@ type (
 )
 
 func (f RequestFunc) Call() {{.Type "C"}} { return f() }
+
 func (f RequestFunc) Marshaler() encoding.Marshaler {
-	call := f()
 	// avoid returning ({{.Type "C"}})(nil) for interface type
-	if call != nil {
+	if call := f(); call != nil {
 		return call
 	}
 	return nil
 }
 
-func (f RequestStreamingFunc) Call() {{.Type "C"}} { return f() }
+func (f RequestStreamingFunc) Push(c ...{{.Type "C"}}) RequestStreamingFunc { return Push(f, c...) }
+
 func (f RequestStreamingFunc) Marshaler() encoding.Marshaler {
-	call := f()
 	// avoid returning ({{.Type "C"}})(nil) for interface type
-	if call != nil {
+	if call := f(); call != nil {
 		return call
 	}
 	return nil
 }
+
 func (f RequestStreamingFunc) IsStreaming() {}
 
+func (f RequestStreamingFunc) Call() {{.Type "C"}} { return f() }
+
+// Push prepends one or more calls onto a request stream. If no calls are given then the original stream is returned.
+func Push(r RequestStreaming, c ...{{.Type "C"}}) RequestStreamingFunc {
+	if len(c) == 0 {
+		return r.Call
+	}
+	var forward int32
+	return func() {{.Type "C"}} {
+		if atomic.LoadInt32(&forward) == 1 {
+			return Push(r, c[1:]...).Call()
+		}
+		atomic.StoreInt32(&forward, 1)
+		return c[0]
+	}
+}
+
+// Empty generates a stream that always returns nil.
+func Empty() RequestStreamingFunc { return func() {{.Type "C"}} { return nil } }
+
 var (
 	_ = Request(RequestFunc(nil))
 	_ = RequestStreaming(RequestStreamingFunc(nil))
@@ -95,11 +117,10 @@ func FromChan(ch <-chan {{.Type "C"}}) RequestStreamingFunc {
 		return func() {{.Type "C"}} { return nil }
 	}
 	return func() {{.Type "C"}} {
-		m, ok := <-ch
-		if !ok {
-			return nil
+		if m, ok := <-ch; ok {
+			return m
 		}
-		return m
+		return nil
 	}
 }
 
@@ -108,13 +129,21 @@ func (f SenderFunc) Send(ctx context.Context, r Request) (mesos.Response, error)
 	return f(ctx, r)
 }
 
+// IgnoreResponse generates a sender that closes any non-nil response received by Mesos.
+func IgnoreResponse(s Sender) SenderFunc {
+	return func(ctx context.Context, r Request) (mesos.Response, error) {
+		resp, err := s.Send(ctx, r)
+		if resp != nil {
+			resp.Close()
+		}
+		return nil, err
+	}
+}
+
 // SendNoData is a convenience func that executes the given Call using the provided Sender
 // and always drops the response data.
-func SendNoData(ctx context.Context, sender Sender, r Request) error {
-	resp, err := sender.Send(ctx, r)
-	if resp != nil {
-		resp.Close()
-	}
-	return err
+func SendNoData(ctx context.Context, sender Sender, r Request) (err error) {
+	_, err = IgnoreResponse(sender).Send(ctx, r)
+	return
 }
 `))
diff --git a/api/v1/lib/httpcli/httpagent/httpagent_generated.go b/api/v1/lib/httpcli/httpagent/httpagent_generated.go
@@ -32,16 +32,9 @@ func NewSender(c *httpcli.Client) calls.Sender {
 
 		var req client.Request
 
-		switch r.(type) {
+		switch r := r.(type) {
 		case calls.RequestStreaming:
-			first := true
-			req = calls.RequestStreamingFunc(func() *agent.Call {
-				if first {
-					first = false
-					return obj
-				}
-				return r.Call()
-			})
+			req = calls.Push(r, obj)
 		default:
 			req = calls.NonStreaming(obj)
 		}

diff --git a/api/v1/lib/httpcli/httpmaster/httpmaster_generated.go b/api/v1/lib/httpcli/httpmaster/httpmaster_generated.go
@@ -32,16 +32,9 @@ func NewSender(c *httpcli.Client) calls.Sender {
 
 		var req client.Request
 
-		switch r.(type) {
+		switch r := r.(type) {
 		case calls.RequestStreaming:
-			first := true
-			req = calls.RequestStreamingFunc(func() *master.Call {
-				if first {
-					first = false
-					return obj
-				}
-				return r.Call()
-			})
+			req = calls.Push(r, obj)
 		default:
 			req = calls.NonStreaming(obj)
 		}

diff --git a/api/v1/lib/master/calls/calls_generated.go b/api/v1/lib/master/calls/calls_generated.go
@@ -5,6 +5,7 @@ package calls
 
 import (
 	"context"
+	"sync/atomic"
 
 	"github.com/mesos/mesos-go/api/v1/lib"
 	"github.com/mesos/mesos-go/api/v1/lib/encoding"
@@ -43,26 +44,47 @@ type (
 )
 
 func (f RequestFunc) Call() *master.Call { return f() }
+
 func (f RequestFunc) Marshaler() encoding.Marshaler {
-	call := f()
 	// avoid returning (*master.Call)(nil) for interface type
-	if call != nil {
+	if call := f(); call != nil {
 		return call
 	}
 	return nil
 }
 
-func (f RequestStreamingFunc) Call() *master.Call { return f() }
+func (f RequestStreamingFunc) Push(c ...*master.Call) RequestStreamingFunc { return Push(f, c...) }
+
 func (f RequestStreamingFunc) Marshaler() encoding.Marshaler {
-	call := f()
 	// avoid returning (*master.Call)(nil) for interface type
-	if call != nil {
+	if call := f(); call != nil {
 		return call
 	}
 	return nil
 }
+
 func (f RequestStreamingFunc) IsStreaming() {}
 
+func (f RequestStreamingFunc) Call() *master.Call { return f() }
+
+// Push prepends one or more calls onto a request stream. If no calls are given then the original stream is returned.
+func Push(r RequestStreaming, c ...*master.Call) RequestStreamingFunc {
+	if len(c) == 0 {
+		return r.Call
+	}
+	var forward int32
+	return func() *master.Call {
+		if atomic.LoadInt32(&forward) == 1 {
+			return Push(r, c[1:]...).Call()
+		}
+		atomic.StoreInt32(&forward, 1)
+		return c[0]
+	}
+}
+
+// Empty generates a stream that always returns nil.
+func Empty() RequestStreamingFunc { return func() *master.Call { return nil } }
+
 var (
 	_ = Request(RequestFunc(nil))
 	_ = RequestStreaming(RequestStreamingFunc(nil))
@@ -80,11 +102,10 @@ func FromChan(ch <-chan *master.Call) RequestStreamingFunc {
 		return func() *master.Call { return nil }
 	}
 	return func() *master.Call {
-		m, ok := <-ch
-		if !ok {
-			return nil
+		if m, ok := <-ch; ok {
+			return m
 		}
-		return m
+		return nil
 	}
 }
 
@@ -93,12 +114,20 @@ func (f SenderFunc) Send(ctx context.Context, r Request) (mesos.Response, error)
 	return f(ctx, r)
 }
 
+// IgnoreResponse generates a sender that closes any non-nil response received by Mesos.
+func IgnoreResponse(s Sender) SenderFunc {
+	return func(ctx context.Context, r Request) (mesos.Response, error) {
+		resp, err := s.Send(ctx, r)
+		if resp != nil {
+			resp.Close()
+		}
+		return nil, err
+	}
+}
+
 // SendNoData is a convenience func that executes the given Call using the provided Sender
 // and always drops the response data.
-func SendNoData(ctx context.Context, sender Sender, r Request) error {
-	resp, err := sender.Send(ctx, r)
-	if resp != nil {
-		resp.Close()
-	}
-	return err
+func SendNoData(ctx context.Context, sender Sender, r Request) (err error) {
+	_, err = IgnoreResponse(sender).Send(ctx, r)
+	return
 }