Add option to pass response to end user

runner/options.go

@@ -5,6 +5,7 @@ import (
 	"crypto/x509"
 	"encoding/json"
 	"fmt"
+	"github.com/gogo/protobuf/proto"
 	"io"
 	"io/ioutil"
 	"math"
@@ -126,12 +127,14 @@ type RunConfig struct {
 	log    Logger
 
 	// misc
-	name        string
-	cpus        int
-	tags        []byte
-	skipFirst   int
-	countErrors bool
-	recvMsgFunc StreamRecvMsgInterceptFunc
+	name             string
+	cpus             int
+	tags             []byte
+	skipFirst        int
+	countErrors      bool
+	recvMsgFunc      StreamRecvMsgInterceptFunc
+	responsesChannel *chan proto.Message
+	stopChan         *chan bool
 }
 
 // Option controls some aspect of run
@@ -142,14 +145,16 @@ func NewConfig(call, host string, options ...Option) (*RunConfig, error) {
 
 	// init with defaults
 	c := &RunConfig{
-		n:            200,
-		c:            50,
-		nConns:       1,
-		timeout:      time.Duration(20 * time.Second),
-		dialTimeout:  time.Duration(10 * time.Second),
-		cpus:         runtime.GOMAXPROCS(-1),
-		zstop:        "close",
-		loadSchedule: ScheduleConst,
+		n:                200,
+		c:                50,
+		nConns:           1,
+		timeout:          time.Duration(20 * time.Second),
+		dialTimeout:      time.Duration(10 * time.Second),
+		cpus:             runtime.GOMAXPROCS(-1),
+		zstop:            "close",
+		loadSchedule:     ScheduleConst,
+		responsesChannel: nil,
+		stopChan:         nil,
 	}
 
 	// apply options
@@ -383,6 +388,14 @@ func WithSkipTLSVerify(skip bool) Option {
 	}
 }
 
+func WithResponsesChannel(channel *chan proto.Message, stopChan *chan bool) Option {
+	return func(o *RunConfig) error {
+		o.responsesChannel = channel
+		o.stopChan = stopChan
+		return nil
+	}
+}
+
 // WithTotalRequests specifies the N (number of total requests) setting
 //	WithTotalRequests(1000)
 func WithTotalRequests(n uint) Option {

runner/requester.go

@@ -196,6 +196,9 @@ func (b *Requester) Run() (*Report, error) {
 func (b *Requester) Stop(reason StopReason) {
 
 	b.stopCh <- true
+	if *b.config.stopChan != nil {
+		*b.config.stopChan <- true
+	}
 
 	b.lock.Lock()
 	b.stopReason = reason
@@ -218,6 +221,9 @@ func (b *Requester) Stop(reason StopReason) {
 // Finish finishes the test run
 func (b *Requester) Finish() *Report {
 	close(b.results)
+	if *b.config.stopChan != nil {
+		*b.config.stopChan <- true
+	}
 	total := time.Since(b.start)
 
 	if b.config.hasLog {

runner/worker.go

@@ -183,6 +183,10 @@ func (w *Worker) makeUnaryRequest(ctx *context.Context, reqMD *metadata.MD, inpu
 			"response", res, "error", resErr)
 	}
 
+	if w.config.responsesChannel != nil {
+		*w.config.responsesChannel <- res
+	}
+
 	return resErr
 }
 
@@ -212,6 +216,10 @@ func (w *Worker) makeClientStreamingRequest(ctx *context.Context,
 				"call", w.mtd.GetFullyQualifiedName(),
 				"response", res, "error", closeErr)
 		}
+
+		if w.config.responsesChannel != nil {
+			*w.config.responsesChannel <- res
+		}
 	}
 
 	performSend := func(payload *dynamic.Message) (bool, error) {
@@ -369,6 +377,10 @@ func (w *Worker) makeServerStreamingRequest(ctx *context.Context, input *dynamic
 				"response", res, "error", err)
 		}
 
+		if w.config.responsesChannel != nil {
+			*w.config.responsesChannel <- res
+		}
+
 		// with any of the cancellation operations we can't just bail
 		// we have to drain the messages until the server gets the cancel and ends their side of the stream
 
@@ -477,6 +489,10 @@ func (w *Worker) makeBidiRequest(ctx *context.Context,
 					"response", res, "error", recvErr)
 			}
 
+			if w.config.responsesChannel != nil {
+				*w.config.responsesChannel <- res
+			}
+
 			if w.streamRecv != nil {
 				if converted, ok := res.(*dynamic.Message); ok {
 					iErr := w.streamRecv(converted, recvErr)