Repository: bojand/ghz
Pull request #254 (2020-12-22 00:57:58)

Additional stream options
Attempt address at least in part #162, #169, #226 and #184.

### CLI

Essentially adds several options for controlling behaviour of client and bidi streaming calls:

#### `--stream-call-duration`

Maximum stream call duration. For client streaming and bidi calls, we'll send messages until this duration expires.

For server streaming calls we will receive message until the duration has expired. Note that in server streaming calls the cancellation will result in call cancelled error.  

Example: `500ms`.

#### `--stream-call-count`

The maximum number of message sends or receives the client will perform in a streaming call before closing the stream and ending the call. For client and bidi streaming calls this dictates the number of messages we will send.

If the data array contains more elements than the count, only data up to the number specified will be used.

If the data array contains fewer elements than the count specified, all the data will be iterated over repeatedly until count limit is reached. 

For server streaming calls we will receive message until the specified count is reached. Note that in server streaming calls the cancellation will result in call cancelled error.  

Examples:

```sh
--stream-call-count=2 -d '[{"name":"Joe"},{"name":"Kate"},{"name":"Sara"}]'
```

Will cause only `[{"name":"Joe"},{"name":"Kate"}]` to be sent. Similarly:

```sh
--stream-call-count=5 -d '[{"name":"Joe"},{"name":"Kate"},{"name":"Sara"}]'
```

Will cause `[{"name":"Joe"},{"name":"Kate"},{"name":"Sara"},{"name":"Joe"},{"name":"Kate"}]` to be sent.

#### `--stream-dynamic-messages`

In streaming calls, regenerate and apply call template data on every message send operation.
This is helpful in combination with template functionality to generate data _for every message sent_ in a streaming call.
For example:

```sh
--stream-dynamic-messages=true --stream-call-count=5 -d '{"name":"{{randomString 8 }}"}'
```

Will result in streaming call with the following data sent:

```json
[{"name":"sKNdMCIb"},{"name":"KLVXDvn1"},{"name":"RJ3knnBh"},{"name":"FTBqQ7nl"},{"name":"FzeMQIWo"}]
```

Contrast that with the default dynamic messages setting turned off; which means the template data will be applied only once for each stream call request, but _not_ for each message sent _in_ the streaming call.

```sh
--stream-call-count=5 -d '{"name":"{{randomString 8 }}"}'
```

Results in the following data sent:

```json
[{"name":"5hL64dd0"},{"name":"5hL64dd0"},{"name":"5hL64dd0"},{"name":"5hL64dd0"},{"name":"5hL64dd0"}]
```

#### `--count-errors`

By default stats for fastest, slowest, average, histogram, and latency distributions only take into account the responses with OK status. This option enabled counting of erroneous (non-OK) responses in stats calculations as well.

### API

In addition to API to support the above options, there are a few Go package API functions introduced in this pull request that are not exposed via CLI options. 

`WithDataProvider` can be used to specify a custom data provider function that's invoked with every call.

```go
WithDataProvider(func(*CallData) ([]*dynamic.Message, error) {
	protoMessage := &helloworld.HelloRequest{Name: "Bob"}
	dynamicMessage, err := dynamic.AsDynamicMessage(protoMessage)
	if err != nil {
		return nil, err
	}
	return []*dynamic.Message{dynamicMessage}, nil
}),
```

The signature is the same for unary or streaming calls. For unary calls the function must return an array with at least 1 value. The first value is always used for unary calls.

Similarly there is a metadata provider function.

```go
WithMetadataProvider(func(*CallData) (*metadata.MD, error) {
	return &metadata.MD{"token": []string{mdv}}, nil
}),
```

`WithStreamRecvMsgIntercept` can be used to add an interceptor function to streaming call that is invoked every time we receive a streaming message. Example usage:

```go
WithStreamRecvMsgIntercept(func(msg *dynamic.Message, err error) error {
	if err == nil && msg != nil {
		reply := &helloworld.HelloReply{}
		convertErr := msg.ConvertTo(reply)
		if convertErr == nil {
			if reply.GetMessage() == "Hello bar" {
				return ErrEndStream
			}
		}
	}
	return nil
})
```

In future release we hope to remove reliance on the `github.com/jhump/protoreflect/dynamic` package for these APIs.


Patch:
diff --git a/README.md b/README.md
index e2b4c9b..9710c92 100644
--- a/README.md
+++ b/README.md
@@ -13,7 +13,7 @@
 [![Donate](https://img.shields.io/badge/Donate-PayPal-green.svg?style=flat-square)](https://www.paypal.me/bojandj)
 [![Buy me a coffee](https://img.shields.io/badge/buy%20me-a%20coffee-orange.svg?style=flat-square)](https://www.buymeacoffee.com/bojand)
 
-Simple [gRPC](http://grpc.io/) benchmarking and load testing tool inspired by [hey](https://github.com/rakyll/hey/) and [grpcurl](https://github.com/fullstorydev/grpcurl).
+[gRPC](http://grpc.io/) benchmarking and load testing tool.
 
 ## Documentation
 
@@ -59,62 +59,66 @@ go build .
 usage: ghz [<flags>] [<host>]
 
 Flags:
-  -h, --help                   Show context-sensitive help (also try --help-long and --help-man).
-      --config=                Path to the JSON or TOML config file that specifies all the test run settings.
-      --proto=                 The Protocol Buffer .proto file.
-      --protoset=              The compiled protoset file. Alternative to proto. -proto takes precedence.
-      --call=                  A fully-qualified method name in 'package.Service/method' or 'package.Service.Method' format.
-  -i, --import-paths=          Comma separated list of proto import paths. The current working directory and the directory of the protocol buffer file are automatically added to the import list.
-      --cacert=                File containing trusted root certificates for verifying the server.
-      --cert=                  File containing client certificate (public key), to present to the server. Must also provide -key option.
-      --key=                   File containing client private key, to present to the server. Must also provide -cert option.
-      --cname=                 Server name override when validating TLS certificate - useful for self signed certs.
-      --skipTLS                Skip TLS client verification of the server's certificate chain and host name.
-      --insecure               Use plaintext and insecure connection.
-      --authority=             Value to be used as the :authority pseudo-header. Only works if -insecure is used.
-      --async                  Make requests asynchronous as soon as possible. Does not wait for request to finish before sending next one.
-  -r, --rps=0                  Requests per second (RPS) rate limit for constant load schedule. Default is no rate limit.
-      --load-schedule="const"  Specifies the load schedule. Options are const, step, or line. Default is const.
-      --load-start=0           Specifies the RPS load start value for step or line schedules.
-      --load-step=0            Specifies the load step value or slope value.
-      --load-end=0             Specifies the load end value for step or line load schedules.
-      --load-step-duration=0   Specifies the load step duration value for step load schedule.
-      --load-max-duration=0    Specifies the max load duration value for step or line load schedule.
-  -c, --concurrency=50         Number of request workers to run concurrently for const concurrency schedule. Default is 50.
+  -h, --help                     Show context-sensitive help (also try --help-long and --help-man).
+      --config=                  Path to the JSON or TOML config file that specifies all the test run settings.
+      --proto=                   The Protocol Buffer .proto file.
+      --protoset=                The compiled protoset file. Alternative to proto. -proto takes precedence.
+      --call=                    A fully-qualified method name in 'package.Service/method' or 'package.Service.Method' format.
+  -i, --import-paths=            Comma separated list of proto import paths. The current working directory and the directory of the protocol buffer file are automatically added to the import list.
+      --cacert=                  File containing trusted root certificates for verifying the server.
+      --cert=                    File containing client certificate (public key), to present to the server. Must also provide -key option.
+      --key=                     File containing client private key, to present to the server. Must also provide -cert option.
+      --cname=                   Server name override when validating TLS certificate - useful for self signed certs.
+      --skipTLS                  Skip TLS client verification of the server's certificate chain and host name.
+      --insecure                 Use plaintext and insecure connection.
+      --authority=               Value to be used as the :authority pseudo-header. Only works if -insecure is used.
+      --async                    Make requests asynchronous as soon as possible. Does not wait for request to finish before sending next one.
+  -r, --rps=0                    Requests per second (RPS) rate limit for constant load schedule. Default is no rate limit.
+      --load-schedule="const"    Specifies the load schedule. Options are const, step, or line. Default is const.
+      --load-start=0             Specifies the RPS load start value for step or line schedules.
+      --load-step=0              Specifies the load step value or slope value.
+      --load-end=0               Specifies the load end value for step or line load schedules.
+      --load-step-duration=0     Specifies the load step duration value for step load schedule.
+      --load-max-duration=0      Specifies the max load duration value for step or line load schedule.
+  -c, --concurrency=50           Number of request workers to run concurrently for const concurrency schedule. Default is 50.
       --concurrency-schedule="const"
-                               Concurrency change schedule. Options are const, step, or line. Default is const.
-      --concurrency-start=0    Concurrency start value for step and line concurrency schedules.
-      --concurrency-end=0      Concurrency end value for step and line concurrency schedules.
-      --concurrency-step=1     Concurrency step / slope value for step and line concurrency schedules.
+                                 Concurrency change schedule. Options are const, step, or line. Default is const.
+      --concurrency-start=0      Concurrency start value for step and line concurrency schedules.
+      --concurrency-end=0        Concurrency end value for step and line concurrency schedules.
+      --concurrency-step=1       Concurrency step / slope value for step and line concurrency schedules.
       --concurrency-step-duration=0
-                               Specifies the concurrency step duration value for step concurrency schedule.
+                                 Specifies the concurrency step duration value for step concurrency schedule.
       --concurrency-max-duration=0
-                               Specifies the max concurrency adjustment duration value for step or line concurrency schedule.
-  -n, --total=200              Number of requests to run. Default is 200.
-  -t, --timeout=20s            Timeout for each request. Default is 20s, use 0 for infinite.
-  -z, --duration=0             Duration of application to send requests. When duration is reached, application stops and exits. If duration is specified, n is ignored. Examples: -z 10s -z 3m.
-  -x, --max-duration=0         Maximum duration of application to send requests with n setting respected. If duration is reached before n requests are completed, application stops and exits. Examples: -x 10s -x 3m.
-      --duration-stop="close"  Specifies how duration stop is reported. Options are close, wait or ignore. Default is close.
-  -d, --data=                  The call data as stringified JSON. If the value is '@' then the request contents are read from stdin.
-  -D, --data-file=             File path for call data JSON file. Examples: /home/user/file.json or ./file.json.
-  -b, --binary                 The call data comes as serialized binary message or multiple count-prefixed messages read from stdin.
-  -B, --binary-file=           File path for the call data as serialized binary message or multiple count-prefixed messages.
-  -m, --metadata=              Request metadata as stringified JSON.
-  -M, --metadata-file=         File path for call metadata JSON file. Examples: /home/user/metadata.json or ./metadata.json.
-      --stream-interval=0      Interval for stream requests between message sends.
-      --reflect-metadata=      Reflect metadata as stringified JSON used only for reflection request.
-  -o, --output=                Output path. If none provided stdout is used.
-  -O, --format=                Output format. One of: summary, csv, json, pretty, html, influx-summary, influx-details. Default is summary.
-      --skipFirst=0            Skip the first X requests when doing the results tally.
-      --connections=1          Number of connections to use. Concurrency is distributed evenly among all the connections. Default is 1.
-      --connect-timeout=10s    Connection timeout for the initial connection dial. Default is 10s.
-      --keepalive=0            Keepalive time duration. Only used if present and above 0.
-      --name=                  User specified name for the test.
-      --tags=                  JSON representation of user-defined string tags.
-      --cpus=12                Number of cpu cores to use.
-      --debug=                 The path to debug log file.
-  -e, --enable-compression     Enable Gzip compression on requests.
-  -v, --version                Show application version.
+                                 Specifies the max concurrency adjustment duration value for step or line concurrency schedule.
+  -n, --total=200                Number of requests to run. Default is 200.
+  -t, --timeout=20s              Timeout for each request. Default is 20s, use 0 for infinite.
+  -z, --duration=0               Duration of application to send requests. When duration is reached, application stops and exits. If duration is specified, n is ignored. Examples: -z 10s -z 3m.
+  -x, --max-duration=0           Maximum duration of application to send requests with n setting respected. If duration is reached before n requests are completed, application stops and exits. Examples: -x 10s -x 3m.
+      --duration-stop="close"    Specifies how duration stop is reported. Options are close, wait or ignore. Default is close.
+  -d, --data=                    The call data as stringified JSON. If the value is '@' then the request contents are read from stdin.
+  -D, --data-file=               File path for call data JSON file. Examples: /home/user/file.json or ./file.json.
+  -b, --binary                   The call data comes as serialized binary message or multiple count-prefixed messages read from stdin.
+  -B, --binary-file=             File path for the call data as serialized binary message or multiple count-prefixed messages.
+  -m, --metadata=                Request metadata as stringified JSON.
+  -M, --metadata-file=           File path for call metadata JSON file. Examples: /home/user/metadata.json or ./metadata.json.
+      --stream-interval=0        Interval for stream requests between message sends.
+      --stream-call-duration=0   Duration after which client will close the stream in each streaming call.
+      --stream-call-count=0      Count of messages sent, after which client will close the stream in each streaming call.
+      --stream-dynamic-messages  In streaming calls, regenerate and apply call template data on every message send.
+      --reflect-metadata=        Reflect metadata as stringified JSON used only for reflection request.
+  -o, --output=                  Output path. If none provided stdout is used.
+  -O, --format=                  Output format. One of: summary, csv, json, pretty, html, influx-summary, influx-details. Default is summary.
+      --skipFirst=0              Skip the first X requests when doing the results tally.
+      --count-errors             Count erroneous (non-OK) resoponses in stats calculations.
+      --connections=1            Number of connections to use. Concurrency is distributed evenly among all the connections. Default is 1.
+      --connect-timeout=10s      Connection timeout for the initial connection dial. Default is 10s.
+      --keepalive=0              Keepalive time duration. Only used if present and above 0.
+      --name=                    User specified name for the test.
+      --tags=                    JSON representation of user-defined string tags.
+      --cpus=12                  Number of cpu cores to use.
+      --debug=                   The path to debug log file.
+  -e, --enable-compression       Enable Gzip compression on requests.
+  -v, --version                  Show application version.
 
 Args:
   [<host>]  Host and port to test.
diff --git a/cmd/ghz/main.go b/cmd/ghz/main.go
index 5ef03ac..e496e9e 100644
--- a/cmd/ghz/main.go
+++ b/cmd/ghz/main.go
@@ -184,6 +184,18 @@ var (
 	si      = kingpin.Flag("stream-interval", "Interval for stream requests between message sends.").
 		Default("0").IsSetByUser(&isSISet).Duration()
 
+	isSCSet = false
+	scd     = kingpin.Flag("stream-call-duration", "Duration after which client will close the stream in each streaming call.").
+		Default("0").IsSetByUser(&isSCSet).Duration()
+
+	isSCCSet = false
+	scc      = kingpin.Flag("stream-call-count", "Count of messages sent, after which client will close the stream in each streaming call.").
+			Default("0").IsSetByUser(&isSCCSet).Uint()
+
+	isSDMSet = false
+	sdm      = kingpin.Flag("stream-dynamic-messages", "In streaming calls, regenerate and apply call template data on every message send.").
+			Default("false").IsSetByUser(&isSDMSet).Bool()
+
 	isRMDSet = false
 	rmd      = kingpin.Flag("reflect-metadata", "Reflect metadata as stringified JSON used only for reflection request.").
 			PlaceHolder(" ").IsSetByUser(&isRMDSet).String()
@@ -201,6 +213,10 @@ var (
 	skipFirst      = kingpin.Flag("skipFirst", "Skip the first X requests when doing the results tally.").
 			Default("0").IsSetByUser(&isSkipFirstSet).Uint()
 
+	isCESet     = false
+	countErrors = kingpin.Flag("count-errors", "Count erroneous (non-OK) resoponses in stats calculations.").
+			Default("false").IsSetByUser(&isCESet).Bool()
+
 	// Connection
 	isConnSet = false
 	conns     = kingpin.Flag("connections", "Number of connections to use. Concurrency is distributed evenly among all the connections. Default is 1.").
@@ -426,6 +442,9 @@ func createConfigFromArgs(cfg *runner.Config) error {
 	cfg.Metadata = metadata
 	cfg.MetadataPath = *mdPath
 	cfg.SI = runner.Duration(*si)
+	cfg.StreamCallDuration = runner.Duration(*scd)
+	cfg.StreamCallCount = *scc
+	cfg.StreamDynamicMessages = *sdm
 	cfg.Output = *output
 	cfg.Format = *format
 	cfg.ImportPaths = iPaths
@@ -451,6 +470,7 @@ func createConfigFromArgs(cfg *runner.Config) error {
 	cfg.CEnd = *cEnd
 	cfg.CStepDuration = runner.Duration(*cStepDuration)
 	cfg.CMaxDuration = runner.Duration(*cMaxDuration)
+	cfg.CountErrors = *countErrors
 
 	return nil
 }
@@ -492,10 +512,6 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.SkipTLSVerify = src.SkipTLSVerify
 	}
 
-	if isSkipFirstSet {
-		dest.SkipFirst = src.SkipFirst
-	}
-
 	if isInsecSet {
 		dest.Insecure = src.Insecure
 	}
@@ -508,6 +524,14 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.CName = src.CName
 	}
 
+	if isSkipFirstSet {
+		dest.SkipFirst = src.SkipFirst
+	}
+
+	if isCESet {
+		dest.CountErrors = src.CountErrors
+	}
+
 	// run
 
 	if isNSet {
@@ -562,6 +586,18 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.SI = src.SI
 	}
 
+	if isSCSet {
+		dest.StreamCallDuration = src.StreamCallDuration
+	}
+
+	if isSCCSet {
+		dest.StreamCallCount = src.StreamCallCount
+	}
+
+	if isSDMSet {
+		dest.StreamDynamicMessages = src.StreamDynamicMessages
+	}
+
 	if isOutputSet {
 		dest.Output = src.Output
 	}
diff --git a/internal/helloworld/greeter_server.go b/internal/helloworld/greeter_server.go
index 73ec43f..8cf6de0 100644
--- a/internal/helloworld/greeter_server.go
+++ b/internal/helloworld/greeter_server.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	context "golang.org/x/net/context"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
 )
 
@@ -30,17 +31,20 @@ var Bidi CallType = "bidi"
 
 // Greeter implements the GreeterServer for tests
 type Greeter struct {
-	streamData []*HelloReply
+	StreamData []*HelloReply
 
 	Stats *HWStatsHandler
 
 	mutex      *sync.RWMutex
 	callCounts map[CallType]int
 	calls      map[CallType][][]*HelloRequest
+
+	sendMutex  *sync.RWMutex
+	sendCounts map[CallType]map[int]int
 }
 
-func randomSleep() {
-	msCount := rand.Intn(4) + 1
+func randomSleep(max int) {
+	msCount := rand.Intn(max) + 1
 	time.Sleep(time.Millisecond * time.Duration(msCount))
 }
 
@@ -62,12 +66,38 @@ func (s *Greeter) recordMessage(ct CallType, callIdx int, msg *HelloRequest) {
 	s.calls[ct][callIdx] = append(s.calls[ct][callIdx], msg)
 }
 
+func (s *Greeter) recordStreamSendCounter(ct CallType, callIdx int) {
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
+
+	s.sendCounts[ct][callIdx] = s.sendCounts[ct][callIdx] + 1
+}
+
 // SayHello implements helloworld.GreeterServer
 func (s *Greeter) SayHello(ctx context.Context, in *HelloRequest) (*HelloReply, error) {
 	callIdx := s.recordCall(Unary)
-	s.recordMessage(Unary, callIdx, in)
 
-	randomSleep()
+	if in.GetName() == "__record_metadata__" {
+		mdval := ""
+		md, ok := metadata.FromIncomingContext(ctx)
+		if ok {
+			for k, v := range md {
+				if k == "token" {
+					mdval = mdval + k + ":"
+					for _, vv := range v {
+						mdval = mdval + vv
+					}
+				}
+			}
+		}
+
+		newReq := &HelloRequest{Name: in.GetName() + "||" + mdval}
+		s.recordMessage(Unary, callIdx, newReq)
+	} else {
+		s.recordMessage(Unary, callIdx, in)
+	}
+
+	randomSleep(4)
 
 	return &HelloReply{Message: "Hello " + in.Name}, nil
 }
@@ -77,12 +107,14 @@ func (s *Greeter) SayHellos(req *HelloRequest, stream Greeter_SayHellosServer) e
 	callIdx := s.recordCall(ServerStream)
 	s.recordMessage(ServerStream, callIdx, req)
 
-	randomSleep()
-
-	for _, msg := range s.streamData {
+	for _, msg := range s.StreamData {
 		if err := stream.Send(msg); err != nil {
 			return err
 		}
+
+		randomSleep(4)
+
+		s.recordStreamSendCounter(ServerStream, callIdx)
 	}
 
 	return nil
@@ -92,7 +124,7 @@ func (s *Greeter) SayHellos(req *HelloRequest, stream Greeter_SayHellosServer) e
 func (s *Greeter) SayHelloCS(stream Greeter_SayHelloCSServer) error {
 	callIdx := s.recordCall(ClientStream)
 
-	randomSleep()
+	randomSleep(4)
 
 	msgCount := 0
 
@@ -114,8 +146,6 @@ func (s *Greeter) SayHelloCS(stream Greeter_SayHelloCSServer) error {
 func (s *Greeter) SayHelloBidi(stream Greeter_SayHelloBidiServer) error {
 	callIdx := s.recordCall(Bidi)
 
-	randomSleep()
-
 	for {
 		in, err := stream.Recv()
 		if err == io.EOF {
@@ -126,10 +156,13 @@ func (s *Greeter) SayHelloBidi(stream Greeter_SayHelloBidiServer) error {
 		}
 
 		s.recordMessage(Bidi, callIdx, in)
+
 		msg := "Hello " + in.Name
 		if err := stream.Send(&HelloReply{Message: msg}); err != nil {
 			return err
 		}
+
+		s.recordStreamSendCounter(ServerStream, callIdx)
 	}
 }
 
@@ -151,6 +184,14 @@ func (s *Greeter) ResetCounters() {
 
 	s.mutex.Unlock()
 
+	s.sendMutex.Lock()
+	s.sendCounts = make(map[CallType]map[int]int)
+	s.sendCounts[Unary] = make(map[int]int)
+	s.sendCounts[ServerStream] = make(map[int]int)
+	s.sendCounts[ClientStream] = make(map[int]int)
+	s.sendCounts[Bidi] = make(map[int]int)
+	s.sendMutex.Unlock()
+
 	if s.Stats != nil {
 		s.Stats.mutex.Lock()
 		s.Stats.connCount = 0
@@ -181,6 +222,25 @@ func (s *Greeter) GetCalls(key CallType) [][]*HelloRequest {
 	return nil
 }
 
+// GetSendCounts gets the stream send counts
+func (s *Greeter) GetSendCounts(key CallType) map[int]int {
+	s.sendMutex.RLock()
+	defer s.sendMutex.RUnlock()
+
+	val, ok := s.sendCounts[key]
+
+	if ok {
+		cm := map[int]int{}
+		for k, v := range val {
+			cm[k] = v
+		}
+
+		return cm
+	}
+
+	return nil
+}
+
 // GetConnectionCount gets the connection count
 func (s *Greeter) GetConnectionCount() int {
 	return s.Stats.GetConnectionCount()
@@ -195,7 +255,7 @@ func NewGreeter() *Greeter {
 		{Message: "Hello Sara"},
 	}
 
-	greeter := &Greeter{streamData: streamData, mutex: &sync.RWMutex{}}
+	greeter := &Greeter{StreamData: streamData, mutex: &sync.RWMutex{}, sendMutex: &sync.RWMutex{}}
 	greeter.ResetCounters()
 
 	return greeter
diff --git a/runner/calldata.go b/runner/calldata.go
index b7c2320..bda1050 100644
--- a/runner/calldata.go
+++ b/runner/calldata.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"math/rand"
 	"text/template"
+	"text/template/parse"
 	"time"
 
 	"github.com/google/uuid"
@@ -34,7 +35,7 @@ type CallData struct {
 	TimestampUnixNano  int64  // timestamp of the call as unix time in nanoseconds
 	UUID               string // generated UUIDv4 for each call
 
-	templateFuncs template.FuncMap
+	t *template.Template
 }
 
 var tmplFuncMap = template.FuncMap{
@@ -47,8 +48,6 @@ func newCallData(
 	mtd *desc.MethodDescriptor,
 	funcs template.FuncMap,
 	workerID string, reqNum int64) *CallData {
-	now := time.Now()
-	newUUID, _ := uuid.NewRandom()
 
 	fns := make(template.FuncMap, len(funcs)+2)
 	for k, v := range tmplFuncMap {
@@ -61,6 +60,11 @@ func newCallData(
 		}
 	}
 
+	t := template.New("call_template_data").Funcs(fns)
+
+	now := time.Now()
+	newUUID, _ := uuid.NewRandom()
+
 	return &CallData{
 		WorkerID:           workerID,
 		RequestNumber:      reqNum,
@@ -76,18 +80,79 @@ func newCallData(
 		TimestampUnixMilli: now.UnixNano() / 1000000,
 		TimestampUnixNano:  now.UnixNano(),
 		UUID:               newUUID.String(),
-		templateFuncs:      fns,
+		t:                  t,
+	}
+}
+
+// Regenerate generates a new instance of call data from this parent instance
+// The dynamic data like timestamps and UUIDs are re-filled
+func (td *CallData) Regenerate() *CallData {
+	now := time.Now()
+	newUUID, _ := uuid.NewRandom()
+
+	return &CallData{
+		WorkerID:           td.WorkerID,
+		RequestNumber:      td.RequestNumber,
+		FullyQualifiedName: td.FullyQualifiedName,
+		MethodName:         td.MethodName,
+		ServiceName:        td.ServiceName,
+		InputName:          td.InputName,
+		OutputName:         td.OutputName,
+		IsClientStreaming:  td.IsClientStreaming,
+		IsServerStreaming:  td.IsServerStreaming,
+		Timestamp:          now.Format(time.RFC3339),
+		TimestampUnix:      now.Unix(),
+		TimestampUnixMilli: now.UnixNano() / 1000000,
+		TimestampUnixNano:  now.UnixNano(),
+		UUID:               newUUID.String(),
+		t:                  td.t,
 	}
 }
 
 func (td *CallData) execute(data string) (*bytes.Buffer, error) {
-	t := template.Must(template.New("call_template_data").Funcs(td.templateFuncs).Parse(data))
+	t, err := td.t.Parse(data)
+	if err != nil {
+		return nil, err
+	}
+
 	var tpl bytes.Buffer
-	err := t.Execute(&tpl, td)
+	err = t.Execute(&tpl, td)
 	return &tpl, err
 }
 
-func (td *CallData) executeData(data string) ([]byte, error) {
+// This is hacky.
+// See https://golang.org/pkg/text/template/#Template
+// The *parse.Tree field is exported only for use by html/template
+// and should be treated as unexported by all other clients.
+func (td *CallData) hasAction(data string) (bool, error) {
+	t, err := td.t.Parse(data)
+	if err != nil {
+		return false, err
+	}
+
+	hasAction := hasAction(t.Tree.Root)
+	return hasAction, nil
+}
+
+func hasAction(node parse.Node) bool {
+	has := false
+	if node.Type() == parse.NodeAction {
+		return true
+	} else if ln, ok := node.(*parse.ListNode); ok {
+		for _, n := range ln.Nodes {
+			v := hasAction(n)
+			if !has && v {
+				has = true
+				break
+			}
+		}
+	}
+
+	return has
+}
+
+// ExecuteData applies the call data's parsed template and data string and returns the resulting buffer
+func (td *CallData) ExecuteData(data string) ([]byte, error) {
 	if len(data) > 0 {
 		input := []byte(data)
 		tpl, err := td.execute(data)
diff --git a/runner/config.go b/runner/config.go
index dec3c21..2d786b5 100644
--- a/runner/config.go
+++ b/runner/config.go
@@ -50,62 +50,67 @@ func (d *Duration) UnmarshalJSON(text []byte) error {
 
 // MarshalJSON implements encoding JSONMarshaler
 func (d Duration) MarshalJSON() ([]byte, error) {
-	return []byte(time.Duration(d).String()), nil
+	return []byte(`"` + time.Duration(d).String() + `"`), nil
 }
 
 // Config for the run.
+// TODO fix casing and consistency.
 type Config struct {
-	Proto             string            `json:"proto" toml:"proto" yaml:"proto"`
-	Protoset          string            `json:"protoset" toml:"protoset" yaml:"protoset"`
-	Call              string            `json:"call" toml:"call" yaml:"call"`
-	RootCert          string            `json:"cacert" toml:"cacert" yaml:"cacert"`
-	Cert              string            `json:"cert" toml:"cert" yaml:"cert"`
-	Key               string            `json:"key" toml:"key" yaml:"key"`
-	SkipTLSVerify     bool              `json:"skipTLS" toml:"skipTLS" yaml:"skipTLS"`
-	SkipFirst         uint              `json:"skipFirst" toml:"skipFirst" yaml:"skipFirst"`
-	CName             string            `json:"cname" toml:"cname" yaml:"cname"`
-	Authority         string            `json:"authority" toml:"authority" yaml:"authority"`
-	Insecure          bool              `json:"insecure,omitempty" toml:"insecure,omitempty" yaml:"insecure,omitempty"`
-	N                 uint              `json:"total" toml:"total" yaml:"total" default:"200"`
-	Async             bool              `json:"async,omitempty" toml:"async,omitempty" yaml:"async,omitempty"`
-	C                 uint              `json:"concurrency" toml:"concurrency" yaml:"concurrency" default:"50"`
-	CSchedule         string            `json:"concurrency-schedule" toml:"concurrency-schedule" yaml:"concurrency-schedule" default:"const"`
-	CStart            uint              `json:"concurrency-start" toml:"concurrency-start" yaml:"concurrency-start" default:"1"`
-	CEnd              uint              `json:"concurrency-end" toml:"concurrency-end" yaml:"concurrency-end" default:"0"`
-	CStep             int               `json:"concurrency-step" toml:"concurrency-step" yaml:"concurrency-step" default:"0"`
-	CStepDuration     Duration          `json:"concurrency-step-duration" toml:"concurrency-step-duration" yaml:"concurrency-step-duration" default:"0"`
-	CMaxDuration      Duration          `json:"concurrency-max-duration" toml:"concurrency-max-duration" yaml:"concurrency-max-duration" default:"0"`
-	Connections       uint              `json:"connections" toml:"connections" yaml:"connections" default:"1"`
-	RPS               uint              `json:"rps" toml:"rps" yaml:"rps"`
-	Z                 Duration          `json:"duration" toml:"duration" yaml:"duration"`
-	ZStop             string            `json:"duration-stop" toml:"duration-stop" yaml:"duration-stop" default:"close"`
-	X                 Duration          `json:"max-duration" toml:"max-duration" yaml:"max-duration"`
-	Timeout           Duration          `json:"timeout" toml:"timeout" yaml:"timeout" default:"20s"`
-	Data              interface{}       `json:"data,omitempty" toml:"data,omitempty" yaml:"data,omitempty"`
-	DataPath          string            `json:"data-file" toml:"data-file" yaml:"data-file"`
-	BinData           []byte            `json:"-" toml:"-" yaml:"-"`
-	BinDataPath       string            `json:"binary-file" toml:"binary-file" yaml:"binary-file"`
-	Metadata          map[string]string `json:"metadata,omitempty" toml:"metadata,omitempty" yaml:"metadata,omitempty"`
-	MetadataPath      string            `json:"metadata-file" toml:"metadata-file" yaml:"metadata-file"`
-	SI                Duration          `json:"stream-interval" toml:"stream-interval" yaml:"stream-interval"`
-	Output            string            `json:"output" toml:"output" yaml:"output"`
-	Format            string            `json:"format" toml:"format" yaml:"format" default:"summary"`
-	DialTimeout       Duration          `json:"connect-timeout" toml:"connect-timeout" yaml:"connect-timeout" default:"10s"`
-	KeepaliveTime     Duration          `json:"keepalive" toml:"keepalive" yaml:"keepalive"`
-	CPUs              uint              `json:"cpus" toml:"cpus" yaml:"cpus"`
-	ImportPaths       []string          `json:"import-paths,omitempty" toml:"import-paths,omitempty" yaml:"import-paths,omitempty"`
-	Name              string            `json:"name,omitempty" toml:"name,omitempty" yaml:"name,omitempty"`
-	Tags              map[string]string `json:"tags,omitempty" toml:"tags,omitempty" yaml:"tags,omitempty"`
-	ReflectMetadata   map[string]string `json:"reflect-metadata,omitempty" toml:"reflect-metadata,omitempty" yaml:"reflect-metadata,omitempty"`
-	Debug             string            `json:"debug,omitempty" toml:"debug,omitempty" yaml:"debug,omitempty"`
-	Host              string            `json:"host" toml:"host" yaml:"host"`
-	EnableCompression bool              `json:"enable-compression,omitempty" toml:"enable-compression,omitempty" yaml:"enable-compression,omitempty"`
-	LoadSchedule      string            `json:"load-schedule" toml:"load-schedule" yaml:"load-schedule" default:"const"`
-	LoadStart         uint              `json:"load-start" toml:"load-start" yaml:"load-start"`
-	LoadEnd           uint              `json:"load-end" toml:"load-end" yaml:"load-end"`
-	LoadStep          int               `json:"load-step" toml:"load-step" yaml:"load-step"`
-	LoadStepDuration  Duration          `json:"load-step-duration" toml:"load-step-duration" yaml:"load-step-duration"`
-	LoadMaxDuration   Duration          `json:"load-max-duration" toml:"load-max-duration" yaml:"load-max-duration"`
+	Proto                 string            `json:"proto" toml:"proto" yaml:"proto"`
+	Protoset              string            `json:"protoset" toml:"protoset" yaml:"protoset"`
+	Call                  string            `json:"call" toml:"call" yaml:"call"`
+	RootCert              string            `json:"cacert" toml:"cacert" yaml:"cacert"`
+	Cert                  string            `json:"cert" toml:"cert" yaml:"cert"`
+	Key                   string            `json:"key" toml:"key" yaml:"key"`
+	CountErrors           bool              `json:"count-errors" toml:"count-errors" yaml:"count-errors"`
+	SkipTLSVerify         bool              `json:"skipTLS" toml:"skipTLS" yaml:"skipTLS"`
+	SkipFirst             uint              `json:"skipFirst" toml:"skipFirst" yaml:"skipFirst"`
+	CName                 string            `json:"cname" toml:"cname" yaml:"cname"`
+	Authority             string            `json:"authority" toml:"authority" yaml:"authority"`
+	Insecure              bool              `json:"insecure,omitempty" toml:"insecure,omitempty" yaml:"insecure,omitempty"`
+	N                     uint              `json:"total" toml:"total" yaml:"total" default:"200"`
+	Async                 bool              `json:"async,omitempty" toml:"async,omitempty" yaml:"async,omitempty"`
+	C                     uint              `json:"concurrency" toml:"concurrency" yaml:"concurrency" default:"50"`
+	CSchedule             string            `json:"concurrency-schedule" toml:"concurrency-schedule" yaml:"concurrency-schedule" default:"const"`
+	CStart                uint              `json:"concurrency-start" toml:"concurrency-start" yaml:"concurrency-start" default:"1"`
+	CEnd                  uint              `json:"concurrency-end" toml:"concurrency-end" yaml:"concurrency-end" default:"0"`
+	CStep                 int               `json:"concurrency-step" toml:"concurrency-step" yaml:"concurrency-step" default:"0"`
+	CStepDuration         Duration          `json:"concurrency-step-duration" toml:"concurrency-step-duration" yaml:"concurrency-step-duration" default:"0"`
+	CMaxDuration          Duration          `json:"concurrency-max-duration" toml:"concurrency-max-duration" yaml:"concurrency-max-duration" default:"0"`
+	Connections           uint              `json:"connections" toml:"connections" yaml:"connections" default:"1"`
+	RPS                   uint              `json:"rps" toml:"rps" yaml:"rps"`
+	Z                     Duration          `json:"duration" toml:"duration" yaml:"duration"`
+	ZStop                 string            `json:"duration-stop" toml:"duration-stop" yaml:"duration-stop" default:"close"`
+	X                     Duration          `json:"max-duration" toml:"max-duration" yaml:"max-duration"`
+	Timeout               Duration          `json:"timeout" toml:"timeout" yaml:"timeout" default:"20s"`
+	Data                  interface{}       `json:"data,omitempty" toml:"data,omitempty" yaml:"data,omitempty"`
+	DataPath              string            `json:"data-file" toml:"data-file" yaml:"data-file"`
+	BinData               []byte            `json:"-" toml:"-" yaml:"-"`
+	BinDataPath           string            `json:"binary-file" toml:"binary-file" yaml:"binary-file"`
+	Metadata              map[string]string `json:"metadata,omitempty" toml:"metadata,omitempty" yaml:"metadata,omitempty"`
+	MetadataPath          string            `json:"metadata-file" toml:"metadata-file" yaml:"metadata-file"`
+	SI                    Duration          `json:"stream-interval" toml:"stream-interval" yaml:"stream-interval"`
+	StreamCallDuration    Duration          `json:"stream-call-duration" toml:"stream-call-duration" yaml:"stream-call-duration"`
+	StreamCallCount       uint              `json:"stream-call-count" toml:"stream-call-count" yaml:"stream-call-count"`
+	StreamDynamicMessages bool              `json:"stream-dynamic-messages" toml:"stream-dynamic-messages" yaml:"stream-dynamic-messages"`
+	Output                string            `json:"output" toml:"output" yaml:"output"`
+	Format                string            `json:"format" toml:"format" yaml:"format" default:"summary"`
+	DialTimeout           Duration          `json:"connect-timeout" toml:"connect-timeout" yaml:"connect-timeout" default:"10s"`
+	KeepaliveTime         Duration          `json:"keepalive" toml:"keepalive" yaml:"keepalive"`
+	CPUs                  uint              `json:"cpus" toml:"cpus" yaml:"cpus"`
+	ImportPaths           []string          `json:"import-paths,omitempty" toml:"import-paths,omitempty" yaml:"import-paths,omitempty"`
+	Name                  string            `json:"name,omitempty" toml:"name,omitempty" yaml:"name,omitempty"`
+	Tags                  map[string]string `json:"tags,omitempty" toml:"tags,omitempty" yaml:"tags,omitempty"`
+	ReflectMetadata       map[string]string `json:"reflect-metadata,omitempty" toml:"reflect-metadata,omitempty" yaml:"reflect-metadata,omitempty"`
+	Debug                 string            `json:"debug,omitempty" toml:"debug,omitempty" yaml:"debug,omitempty"`
+	Host                  string            `json:"host" toml:"host" yaml:"host"`
+	EnableCompression     bool              `json:"enable-compression,omitempty" toml:"enable-compression,omitempty" yaml:"enable-compression,omitempty"`
+	LoadSchedule          string            `json:"load-schedule" toml:"load-schedule" yaml:"load-schedule" default:"const"`
+	LoadStart             uint              `json:"load-start" toml:"load-start" yaml:"load-start"`
+	LoadEnd               uint              `json:"load-end" toml:"load-end" yaml:"load-end"`
+	LoadStep              int               `json:"load-step" toml:"load-step" yaml:"load-step"`
+	LoadStepDuration      Duration          `json:"load-step-duration" toml:"load-step-duration" yaml:"load-step-duration"`
+	LoadMaxDuration       Duration          `json:"load-max-duration" toml:"load-max-duration" yaml:"load-max-duration"`
 }
 
 func checkData(data interface{}) error {
diff --git a/runner/data.go b/runner/data.go
index 32f0606..e521dc2 100644
--- a/runner/data.go
+++ b/runner/data.go
@@ -2,16 +2,237 @@ package runner
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"strings"
+	"sync"
 
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/golang/protobuf/proto"
 	"github.com/jhump/protoreflect/desc"
 	"github.com/jhump/protoreflect/dynamic"
+	"google.golang.org/grpc/metadata"
 )
 
+// TODO move to own pacakge?
+// TODO add tests
+// TODO expose public API utilizing only proto API and not dynamic
+
+// ErrEndStream is a signal from message providers that worker should close the stream
+// It should not be used for erronous states
+var ErrEndStream = errors.New("ending stream")
+
+// ErrLastMessage is a signal from message providers that the returned payload is the last one of the stream
+// This is optional but encouraged for optimized performance
+// Message payload returned along with this error must be valid and may not be nil
+var ErrLastMessage = errors.New("last message")
+
+// DataProviderFunc is the interface for providing data for calls
+// For unary and server streaming calls it should return an array with a single element
+// For client and bidi streaming calls it should return an array of messages to be used
+type DataProviderFunc func(*CallData) ([]*dynamic.Message, error)
+
+// MetadataProviderFunc is the interface for providing metadadata for calls
+type MetadataProviderFunc func(*CallData) (*metadata.MD, error)
+
+// StreamMessageProviderFunc is the interface for providing a message for every message send in the course of a streaming call
+type StreamMessageProviderFunc func(*CallData) (*dynamic.Message, error)
+
+// StreamRecvMsgInterceptFunc is an interface for function invoked when we receive a stream message
+// Clients can return ErrEndStream to end the call early
+type StreamRecvMsgInterceptFunc func(*dynamic.Message, error) error
+
+type dataProvider struct {
+	binary   bool
+	data     []byte
+	mtd      *desc.MethodDescriptor
+	dataFunc BinaryDataFunc
+
+	arrayJSONData []string
+	hasActions    bool
+
+	// cached messages only for binary
+	mutex          sync.RWMutex
+	cachedMessages []*dynamic.Message
+}
+
+type mdProvider struct {
+	metadata []byte
+}
+
+func newDataProvider(mtd *desc.MethodDescriptor,
+	binary bool, dataFunc BinaryDataFunc, data []byte) (*dataProvider, error) {
+
+	dp := dataProvider{
+		binary:         binary,
+		dataFunc:       dataFunc,
+		mtd:            mtd,
+		data:           data,
+		cachedMessages: nil,
+	}
+
+	// fill in JSON string array data for optimization for non client-streaming
+	var err error
+	dp.arrayJSONData = nil
+	if !dp.binary {
+		if strings.IndexRune(string(data), '[') == 0 { // it's an array
+			var dat []map[string]interface{}
+			if err := json.Unmarshal(data, &dat); err != nil {
+				return nil, err
+			}
+
+			dp.arrayJSONData = make([]string, len(dat))
+			for i, d := range dat {
+				var strd []byte
+				if strd, err = json.Marshal(d); err != nil {
+					return nil, err
+				}
+
+				dp.arrayJSONData[i] = string(strd)
+			}
+		}
+	}
+
+	// Test if we can preseed data
+	ctd := newCallData(mtd, nil, "", 0)
+	ha, err := ctd.hasAction(string(dp.data))
+	if err != nil {
+		return nil, err
+	}
+
+	dp.hasActions = ha
+
+	if !ha {
+		if len(dp.arrayJSONData) > 0 {
+			dp.mutex.Lock()
+			dp.cachedMessages = make([]*dynamic.Message, len(dp.arrayJSONData))
+			dp.mutex.Unlock()
+		}
+
+		_, err := dp.getDataForCall(ctd)
+		if err != nil {
+			return nil, err
+		}
+
+	}
+
+	return &dp, nil
+}
+
+func (dp *dataProvider) getDataForCall(ctd *CallData) ([]*dynamic.Message, error) {
+	var inputs []*dynamic.Message
+	var err error
+
+	// try the optimized path for JSON data for non client-streaming
+	if !dp.binary && !dp.mtd.IsClientStreaming() && len(dp.arrayJSONData) > 0 {
+		indx := int(ctd.RequestNumber % int64(len(dp.arrayJSONData))) // we want to start from inputs[0] so dec reqNum
+
+		if inputs, err = dp.getMessages(ctd, indx, []byte(dp.arrayJSONData[indx])); err != nil {
+			return nil, err
+		}
+	} else if inputs, err = dp.getMessages(ctd, -1, dp.data); err != nil {
+		return nil, err
+	}
+
+	if !dp.mtd.IsClientStreaming() {
+		inputIdx := int(ctd.RequestNumber % int64(len(inputs)))
+		unaryInput := inputs[inputIdx]
+
+		return []*dynamic.Message{unaryInput}, nil
+	}
+
+	return inputs, nil
+}
+
+func (dp *dataProvider) getMessages(ctd *CallData, i int, inputData []byte) ([]*dynamic.Message, error) {
+	var inputs []*dynamic.Message
+	var err error
+
+	dp.mutex.RLock()
+	if dp.cachedMessages != nil {
+		if (i < 0 && len(dp.cachedMessages) > 0 && dp.cachedMessages[0] != nil) ||
+			(i >= 0 && i < len(dp.cachedMessages) && dp.cachedMessages[i] != nil) {
+			defer dp.mutex.RUnlock()
+			return dp.cachedMessages, nil
+		}
+	}
+	dp.mutex.RUnlock()
+
+	if !dp.binary {
+		data := inputData
+
+		if dp.hasActions {
+			cdata, err := ctd.ExecuteData(string(inputData))
+			if err != nil {
+				return nil, err
+			}
+			data = cdata
+		}
+
+		inputs, err = createPayloadsFromJSON(string(data), dp.mtd)
+		if err != nil {
+			return nil, err
+		}
+
+		// only cache JSON data if there are no template actions
+		if !dp.hasActions {
+			dp.mutex.Lock()
+			if i < 0 {
+				dp.cachedMessages = inputs
+			} else {
+				if i >= cap(dp.cachedMessages) {
+					nc := make([]*dynamic.Message, len(dp.cachedMessages), (cap(dp.cachedMessages) + i + 1))
+					copy(nc, dp.cachedMessages)
+					dp.cachedMessages = nc
+				}
+
+				if i < len(dp.cachedMessages) {
+					dp.cachedMessages[i] = inputs[0]
+				} else {
+					dp.cachedMessages = append(dp.cachedMessages, inputs...)
+				}
+			}
+			dp.mutex.Unlock()
+		}
+	} else {
+		var err error
+		if dp.dataFunc != nil {
+			inputData = dp.dataFunc(dp.mtd, ctd)
+		}
+		inputs, err = createPayloadsFromBin(inputData, dp.mtd)
+		if err != nil {
+			return nil, err
+		}
+		// We only cache in case we don't dynamically change the binary message
+		if dp.dataFunc == nil {
+			dp.mutex.Lock()
+			dp.cachedMessages = inputs
+			dp.mutex.Unlock()
+		}
+	}
+
+	return inputs, nil
+}
+
+func newMetadataProvider(mtd *desc.MethodDescriptor, metadata []byte) (*mdProvider, error) {
+	return &mdProvider{metadata: metadata}, nil
+}
+
+func (dp *mdProvider) getMetadataForCall(ctd *CallData) (*metadata.MD, error) {
+	mdMap, err := ctd.executeMetadata(string(dp.metadata))
+	if err != nil {
+		return nil, err
+	}
+
+	if len(mdMap) > 0 {
+		md := metadata.New(mdMap)
+		return &md, nil
+	}
+
+	return &metadata.MD{}, nil
+}
+
 // creates a message from a map
 // marshal to JSON then use jsonpb to marshal to message
 // this way we follow protobuf more closely and allow camelCase properties.
@@ -127,3 +348,152 @@ func createPayloadsFromBin(binData []byte, mtd *desc.MethodDescriptor) ([]*dynam
 
 	return createPayloadsFromBinSingleMessage(binData, mtd)
 }
+
+type dynamicMessageProvider struct {
+	mtd           *desc.MethodDescriptor
+	data          []byte
+	arrayJSONData []string
+	arrayLen      uint
+
+	streamCallCount uint
+	counter         uint
+	indexCounter    uint
+}
+
+func newDynamicMessageProvider(mtd *desc.MethodDescriptor, data []byte, streamCallCount uint) (*dynamicMessageProvider, error) {
+	mp := dynamicMessageProvider{
+		mtd:             mtd,
+		data:            data,
+		arrayJSONData:   nil,
+		streamCallCount: streamCallCount,
+	}
+
+	var err error
+
+	if strings.IndexRune(string(data), '[') == 0 { // it's an array
+		var dat []map[string]interface{}
+		if err := json.Unmarshal(data, &dat); err != nil {
+			return nil, err
+		}
+
+		mp.arrayJSONData = make([]string, len(dat))
+		for i, d := range dat {
+			var strd []byte
+			if strd, err = json.Marshal(d); err != nil {
+				return nil, err
+			}
+
+			mp.arrayJSONData[i] = string(strd)
+		}
+	}
+
+	mp.arrayLen = uint(len(mp.arrayJSONData))
+
+	// Test if we have actions
+	ctd := newCallData(mtd, nil, "", 0)
+	ha, err := ctd.hasAction(string(mp.data))
+	if err != nil {
+		return nil, err
+	}
+
+	if !ha {
+		return nil, errors.New("default message provider cannot be used with static data")
+	}
+
+	return &mp, nil
+}
+
+func (m *dynamicMessageProvider) GetStreamMessage(parentCallData *CallData) (*dynamic.Message, error) {
+	isLast := false
+	if m.streamCallCount > 0 {
+		if m.counter >= m.streamCallCount {
+			return nil, ErrEndStream
+		} else if m.counter >= m.arrayLen {
+			m.indexCounter = 0
+		}
+
+		isLast = m.counter == m.streamCallCount-1
+	} else if m.counter >= m.arrayLen {
+		return nil, ErrEndStream
+	} else if m.counter == m.arrayLen-1 {
+		isLast = true
+	}
+
+	ctd := parentCallData.Regenerate()
+
+	data := string(m.data)
+
+	if m.arrayLen > 0 {
+		if m.counter >= m.arrayLen {
+			m.indexCounter = 0
+		}
+
+		data = m.arrayJSONData[m.indexCounter]
+	}
+
+	buf, err := ctd.ExecuteData(data)
+	if err != nil {
+		return nil, err
+	}
+
+	md := m.mtd.GetInputType()
+	msg := dynamic.NewMessage(md)
+	err = jsonpb.UnmarshalString(string(buf), msg)
+	if err != nil {
+		return nil, fmt.Errorf("Error creating message from data. Data: '%v' Error: %v", data, err.Error())
+	}
+
+	m.counter++
+	m.indexCounter++
+
+	if err == nil && isLast {
+		err = ErrLastMessage
+	}
+
+	return msg, err
+}
+
+type staticMessageProvider struct {
+	inputs          []*dynamic.Message
+	inputLen        uint
+	streamCallCount uint
+	counter         uint
+	indexCounter    uint
+}
+
+func newStaticMessageProvider(streamCallCount uint, inputs []*dynamic.Message) (*staticMessageProvider, error) {
+	return &staticMessageProvider{
+		streamCallCount: streamCallCount,
+		inputs:          inputs,
+		inputLen:        uint(len(inputs)),
+	}, nil
+}
+
+func (m *staticMessageProvider) GetStreamMessage(parentCallData *CallData) (*dynamic.Message, error) {
+	isLast := false
+	if m.streamCallCount > 0 {
+		if m.counter >= m.streamCallCount {
+			return nil, ErrEndStream
+		} else if m.indexCounter == m.inputLen {
+			m.indexCounter = 0
+		}
+
+		isLast = m.counter == m.streamCallCount-1
+	} else if m.counter >= m.inputLen {
+		return nil, ErrEndStream
+	} else if m.counter == m.inputLen-1 {
+		isLast = true
+	}
+
+	payload := m.inputs[m.indexCounter]
+
+	m.counter++
+	m.indexCounter++
+
+	var err error
+	if err == nil && isLast {
+		err = ErrLastMessage
+	}
+
+	return payload, err
+}
diff --git a/runner/options.go b/runner/options.go
index 2e4ebdc..8ef25c1 100644
--- a/runner/options.go
+++ b/runner/options.go
@@ -92,31 +92,40 @@ type RunConfig struct {
 
 	zstop string
 
-	streamInterval time.Duration
-
-	// data
-	data []byte
+	streamInterval        time.Duration
+	streamCallDuration    time.Duration
+	streamCallCount       uint
+	streamDynamicMessages bool
 
 	// lbStrategy
 	lbStrategy string
-	// data func
-	dataFunc BinaryDataFunc
 
-	binary   bool
+	// TODO consolidate these actual value fields to be implemented via provider funcs
+	// data & metadata
+	data     []byte
 	metadata []byte
-	rmd      map[string]string
+	binary   bool
+
+	dataFunc         BinaryDataFunc
+	dataProviderFunc DataProviderFunc
+	mdProviderFunc   MetadataProviderFunc
+
+	funcs template.FuncMap
+
+	// reflection metadata
+	rmd map[string]string
 
 	// debug
 	hasLog bool
 	log    Logger
 
 	// misc
-	name      string
-	cpus      int
-	tags      []byte
-	skipFirst int
-
-	funcs template.FuncMap
+	name        string
+	cpus        int
+	tags        []byte
+	skipFirst   int
+	countErrors bool
+	recvMsgFunc StreamRecvMsgInterceptFunc
 }
 
 // Option controls some aspect of run
@@ -657,6 +666,15 @@ func WithSkipFirst(c uint) Option {
 	}
 }
 
+// WithCountErrors is the count errors option
+func WithCountErrors(v bool) Option {
+	return func(o *RunConfig) error {
+		o.countErrors = v
+
+		return nil
+	}
+}
+
 // WithProtoFile specified proto file path and optionally import paths
 // We will automatically add the proto file path's directory and the current directory
 //	WithProtoFile("greeter.proto", []string{"/home/protos"})
@@ -706,6 +724,33 @@ func WithStreamInterval(d time.Duration) Option {
 	}
 }
 
+// WithStreamCallDuration sets the maximum stream call duration at which point the client will close the stream
+func WithStreamCallDuration(d time.Duration) Option {
+	return func(o *RunConfig) error {
+		o.streamCallDuration = d
+
+		return nil
+	}
+}
+
+// WithStreamCallCount sets the stream close count
+func WithStreamCallCount(c uint) Option {
+	return func(o *RunConfig) error {
+		o.streamCallCount = c
+
+		return nil
+	}
+}
+
+// WithStreamDynamicMessages sets the stream dynamic message generation
+func WithStreamDynamicMessages(v bool) Option {
+	return func(o *RunConfig) error {
+		o.streamDynamicMessages = v
+
+		return nil
+	}
+}
+
 // WithReflectionMetadata specifies the metadata to be used as a map
 // 	md := make(map[string]string)
 // 	md["token"] = "foobar"
@@ -909,6 +954,58 @@ func WithWorkerTicker(ticker load.WorkerTicker) Option {
 	}
 }
 
+// WithStreamRecvMsgIntercept specified the stream receive intercept function
+//
+//	WithStreamRecvMsgIntercept(func(msg *dynamic.Message, err error) error {
+//		if err == nil && msg != nil {
+//			reply := &helloworld.HelloReply{}
+//			convertErr := msg.ConvertTo(reply)
+//			if convertErr == nil {
+//				if reply.GetMessage() == "Hello bar" {
+//					return ErrEndStream
+//				}
+//			}
+//		}
+//		return nil
+//	})
+//
+func WithStreamRecvMsgIntercept(fn StreamRecvMsgInterceptFunc) Option {
+	return func(o *RunConfig) error {
+		o.recvMsgFunc = fn
+
+		return nil
+	}
+}
+
+// WithDataProvider provides custom data provider
+//	WithDataProvider(func(*CallData) ([]*dynamic.Message, error) {
+//		protoMsg := &helloworld.HelloRequest{Name: "Bob"}
+//		dynamicMsg, err := dynamic.AsDynamicMessage(protoMsg)
+//		if err != nil {
+//			return nil, err
+//		}
+//		return []*dynamic.Message{dynamicMsg}, nil
+//	}),
+func WithDataProvider(fn DataProviderFunc) Option {
+	return func(o *RunConfig) error {
+		o.dataProviderFunc = fn
+
+		return nil
+	}
+}
+
+// WithMetadataProvider provides custom metadata provider
+//	WithMetadataProvider(ctd *CallData) (*metadata.MD, error) {
+//		return &metadata.MD{"token": []string{"secret"}}, nil
+//	}),
+func WithMetadataProvider(fn MetadataProviderFunc) Option {
+	return func(o *RunConfig) error {
+		o.mdProviderFunc = fn
+
+		return nil
+	}
+}
+
 func createClientTransportCredentials(skipVerify bool, cacertFile, clientCertFile, clientKeyFile, cname string) (credentials.TransportCredentials, error) {
 	var tlsConf tls.Config
 
@@ -979,6 +1076,9 @@ func fromConfig(cfg *Config) []Option {
 		WithMetadata(cfg.Metadata),
 		WithTags(cfg.Tags),
 		WithStreamInterval(time.Duration(cfg.SI)),
+		WithStreamCallDuration(time.Duration(cfg.StreamCallDuration)),
+		WithStreamCallCount(cfg.StreamCallCount),
+		WithStreamDynamicMessages(cfg.StreamDynamicMessages),
 		WithReflectionMetadata(cfg.ReflectMetadata),
 		WithConnections(cfg.Connections),
 		WithEnableCompression(cfg.EnableCompression),
@@ -996,6 +1096,7 @@ func fromConfig(cfg *Config) []Option {
 		WithConcurrencyStep(cfg.CStep),
 		WithConcurrencyStepDuration(time.Duration(cfg.CStepDuration)),
 		WithConcurrencyDuration(time.Duration(cfg.CMaxDuration)),
+		WithCountErrors(cfg.CountErrors),
 		func(o *RunConfig) error {
 			o.call = cfg.Call
 			return nil
diff --git a/runner/reporter.go b/runner/reporter.go
index 6687d78..9f3c8fd 100644
--- a/runner/reporter.go
+++ b/runner/reporter.go
@@ -23,6 +23,7 @@ type Reporter struct {
 }
 
 // Options represents the request options
+// TODO fix casing and consistency
 type Options struct {
 	Call              string   `json:"call,omitempty"`
 	Host              string   `json:"host,omitempty"`
@@ -71,7 +72,8 @@ type Options struct {
 	CPUs int    `json:"CPUs"`
 	Name string `json:"name,omitempty"`
 
-	SkipFirst uint `json:"skipFirst,omitempty"`
+	SkipFirst   uint `json:"skipFirst,omitempty"`
+	CountErrors bool `json:"count-errors,omitempty"`
 }
 
 // Report holds the data for the full test
@@ -161,6 +163,7 @@ func (r *Reporter) Run() {
 			skipCount++
 			continue
 		}
+
 		errStr := ""
 		r.totalCount++
 		r.totalLatenciesSec += res.duration.Seconds()
@@ -235,10 +238,11 @@ func (r *Reporter) Finalize(stopReason StopReason, total time.Duration) *Report
 		DialTimeout:   r.config.dialTimeout,
 		KeepaliveTime: r.config.keepaliveTime,
 
-		Binary:    r.config.binary,
-		CPUs:      r.config.cpus,
-		Name:      r.config.name,
-		SkipFirst: uint(r.config.skipFirst),
+		Binary:      r.config.binary,
+		CPUs:        r.config.cpus,
+		Name:        r.config.name,
+		SkipFirst:   uint(r.config.skipFirst),
+		CountErrors: r.config.countErrors,
 	}
 
 	_ = json.Unmarshal(r.config.data, &rep.Options.Data)
@@ -255,7 +259,7 @@ func (r *Reporter) Finalize(stopReason StopReason, total time.Duration) *Report
 
 		okLats := make([]float64, 0)
 		for _, d := range r.details {
-			if d.Error == "" {
+			if d.Error == "" || rep.Options.CountErrors {
 				okLats = append(okLats, d.Latency.Seconds())
 			}
 		}
diff --git a/runner/requester.go b/runner/requester.go
index 20874d5..e233bca 100644
--- a/runner/requester.go
+++ b/runner/requester.go
@@ -2,11 +2,9 @@ package runner
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"math"
 	"strconv"
-	"strings"
 	"sync"
 	"time"
 
@@ -19,6 +17,7 @@ import (
 
 	"go.uber.org/multierr"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 
@@ -51,7 +50,8 @@ type Requester struct {
 	stopCh  chan bool
 	start   time.Time
 
-	arrayJSONData []string
+	dataProvider     DataProviderFunc
+	metadataProvider MetadataProviderFunc
 
 	lock       sync.Mutex
 	stopReason StopReason
@@ -122,25 +122,24 @@ func NewRequester(c *RunConfig) (*Requester, error) {
 	// fill in the rest
 	reqr.mtd = mtd
 
-	// fill in JSON string array data for optimization for non client-streaming
-	reqr.arrayJSONData = nil
-	if !c.binary && !reqr.mtd.IsClientStreaming() {
-		if strings.IndexRune(string(c.data), '[') == 0 { // it's an array
-			var dat []map[string]interface{}
-			if err := json.Unmarshal(c.data, &dat); err != nil {
-				return nil, err
-			}
-
-			reqr.arrayJSONData = make([]string, len(dat))
-			for i, d := range dat {
-				var strd []byte
-				if strd, err = json.Marshal(d); err != nil {
-					return nil, err
-				}
+	if c.dataProviderFunc != nil {
+		reqr.dataProvider = c.dataProviderFunc
+	} else {
+		defaultDataProvider, err := newDataProvider(reqr.mtd, c.binary, c.dataFunc, c.data)
+		if err != nil {
+			return nil, err
+		}
+		reqr.dataProvider = defaultDataProvider.getDataForCall
+	}
 
-				reqr.arrayJSONData[i] = string(strd)
-			}
+	if c.mdProviderFunc != nil {
+		reqr.metadataProvider = c.mdProviderFunc
+	} else {
+		defaultMDProvider, err := newMetadataProvider(reqr.mtd, c.metadata)
+		if err != nil {
+			return nil, err
 		}
+		reqr.metadataProvider = defaultMDProvider.getMetadataForCall
 	}
 
 	return reqr, nil
@@ -182,6 +181,7 @@ func (b *Requester) Run() (*Report, error) {
 	err = b.runWorkers(wt, p)
 
 	report := b.Finish()
+
 	b.closeClientConns()
 
 	return report, err
@@ -270,7 +270,14 @@ func (b *Requester) closeClientConns() {
 	}
 
 	for _, cc := range b.conns {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second*10))
+		defer cancel()
+
+		shutdownCh := connectionOnState(ctx, cc, connectivity.Shutdown)
+
 		_ = cc.Close()
+
+		<-shutdownCh
 	}
 
 	b.conns = nil
@@ -366,19 +373,20 @@ func (b *Requester) runWorkers(wt load.WorkerTicker, p load.Pacer) error {
 					}
 
 					if b.config.hasLog {
-						b.config.log.Debugw("Creating worker with ID: "+wID,
-							"workerID", wID, "requests per worker")
+						b.config.log.Debugw("Creating worker with ID: "+wID, "workerID", wID)
 					}
 
 					w := Worker{
-						ticks:         ticks,
-						active:        true,
-						stub:          b.stubs[n],
-						mtd:           b.mtd,
-						config:        b.config,
-						stopCh:        make(chan bool),
-						workerID:      wID,
-						arrayJSONData: b.arrayJSONData,
+						ticks:            ticks,
+						active:           true,
+						stub:             b.stubs[n],
+						mtd:              b.mtd,
+						config:           b.config,
+						stopCh:           make(chan bool),
+						workerID:         wID,
+						dataProvider:     b.dataProvider,
+						metadataProvider: b.metadataProvider,
+						streamRecv:       b.config.recvMsgFunc,
 					}
 
 					wc++ // increment worker id
@@ -464,7 +472,10 @@ func (b *Requester) runWorkers(wt load.WorkerTicker, p load.Pacer) error {
 	<-done
 
 	var err error
-	for i := 0; i < len(b.workers); i++ {
+	wm.Lock()
+	nw := len(b.workers)
+	wm.Unlock()
+	for i := 0; i < nw; i++ {
 		err = multierr.Append(err, <-errC)
 	}
 
@@ -539,3 +550,42 @@ func createPacer(config *RunConfig) load.Pacer {
 
 	return p
 }
+
+func checkState(conn *grpc.ClientConn, states ...connectivity.State) bool {
+	currentState := conn.GetState()
+	for _, s := range states {
+		if currentState == s {
+			return true
+		}
+	}
+
+	return false
+}
+
+func connectionOnState(ctx context.Context, conn *grpc.ClientConn, states ...connectivity.State) <-chan bool {
+
+	stateCh := make(chan bool)
+
+	go func() {
+		defer close(stateCh)
+		if checkState(conn, states...) {
+			stateCh <- true
+			return
+		}
+
+		for {
+			change := conn.WaitForStateChange(ctx, conn.GetState())
+			if !change {
+				stateCh <- checkState(conn, states...)
+				return
+			}
+
+			if checkState(conn, states...) {
+				stateCh <- true
+				return
+			}
+		}
+	}()
+
+	return stateCh
+}
diff --git a/runner/worker.go b/runner/worker.go
index 47d9aad..b647f78 100644
--- a/runner/worker.go
+++ b/runner/worker.go
@@ -2,6 +2,7 @@ package runner
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"time"
@@ -34,11 +35,9 @@ type Worker struct {
 	stopCh   chan bool
 	ticks    <-chan TickValue
 
-	// cached messages only for binary
-	cachedMessages []*dynamic.Message
-
-	// non-binary json optimization
-	arrayJSONData []string
+	dataProvider     DataProviderFunc
+	metadataProvider MetadataProviderFunc
+	streamRecv       StreamRecvMsgInterceptFunc
 }
 
 func (w *Worker) runWorker() error {
@@ -53,7 +52,6 @@ func (w *Worker) runWorker() error {
 			}
 
 			return err
-
 		case tv := <-w.ticks:
 			if w.config.async {
 				g.Go(func() error {
@@ -82,32 +80,17 @@ func (w *Worker) makeRequest(tv TickValue) error {
 
 	ctd := newCallData(w.mtd, w.config.funcs, w.workerID, reqNum)
 
-	var inputs []*dynamic.Message
-	var err error
-
-	// try the optimized path for JSON data for non client-streaming
-	if !w.config.binary && !w.mtd.IsClientStreaming() && len(w.arrayJSONData) > 0 {
-		indx := int(reqNum % int64(len(w.arrayJSONData))) // we want to start from inputs[0] so dec reqNum
-		if inputs, err = w.getMessages(ctd, []byte(w.arrayJSONData[indx])); err != nil {
-			return err
-		}
-	} else {
-		if inputs, err = w.getMessages(ctd, w.config.data); err != nil {
-			return err
-		}
-	}
-
-	mdMap, err := ctd.executeMetadata(string(w.config.metadata))
+	inputs, err := w.dataProvider(ctd)
 	if err != nil {
 		return err
 	}
+	if len(inputs) == 0 {
+		return fmt.Errorf("no data provided for request")
+	}
 
-	var reqMD *metadata.MD
-	if len(mdMap) > 0 {
-		md := metadata.New(mdMap)
-		reqMD = &md
-	} else {
-		reqMD = &metadata.MD{}
+	reqMD, err := w.metadataProvider(ctd)
+	if err != nil {
+		return err
 	}
 
 	if w.config.enableCompression {
@@ -145,18 +128,31 @@ func (w *Worker) makeRequest(tv TickValue) error {
 			"input", inputs, "metadata", reqMD)
 	}
 
-	inputsLen := len(inputs)
-	if inputsLen == 0 {
-		return fmt.Errorf("no data provided for request")
+	unaryInput := inputs[0]
+	var msgProvider StreamMessageProviderFunc
+	if w.mtd.IsClientStreaming() {
+		if w.config.streamDynamicMessages {
+			mp, err := newDynamicMessageProvider(w.mtd, w.config.data, w.config.streamCallCount)
+			if err != nil {
+				return err
+			}
+
+			msgProvider = mp.GetStreamMessage
+		} else {
+			mp, err := newStaticMessageProvider(w.config.streamCallCount, inputs)
+			if err != nil {
+				return err
+			}
+
+			msgProvider = mp.GetStreamMessage
+		}
 	}
-	inputIdx := int(reqNum % int64(inputsLen)) // we want to start from inputs[0] so dec reqNum
-	unaryInput := inputs[inputIdx]
 
 	// RPC errors are handled via stats handler
 	if w.mtd.IsClientStreaming() && w.mtd.IsServerStreaming() {
-		_ = w.makeBidiRequest(&ctx, inputs)
+		_ = w.makeBidiRequest(&ctx, ctd, msgProvider)
 	} else if w.mtd.IsClientStreaming() {
-		_ = w.makeClientStreamingRequest(&ctx, inputs)
+		_ = w.makeClientStreamingRequest(&ctx, ctd, msgProvider)
 	} else if w.mtd.IsServerStreaming() {
 		_ = w.makeServerStreamingRequest(&ctx, unaryInput)
 	} else {
@@ -166,41 +162,6 @@ func (w *Worker) makeRequest(tv TickValue) error {
 	return err
 }
 
-func (w *Worker) getMessages(ctd *CallData, inputData []byte) ([]*dynamic.Message, error) {
-	var inputs []*dynamic.Message
-
-	if w.cachedMessages != nil {
-		return w.cachedMessages, nil
-	}
-
-	if !w.config.binary {
-		data, err := ctd.executeData(string(inputData))
-		if err != nil {
-			return nil, err
-		}
-		inputs, err = createPayloadsFromJSON(string(data), w.mtd)
-		if err != nil {
-			return nil, err
-		}
-		// Json messages are not cached due to templating
-	} else {
-		var err error
-		if w.config.dataFunc != nil {
-			inputData = w.config.dataFunc(w.mtd, ctd)
-		}
-		inputs, err = createPayloadsFromBin(inputData, w.mtd)
-		if err != nil {
-			return nil, err
-		}
-		// We only cache in case we don't dynamically change the binary message
-		if w.config.dataFunc == nil {
-			w.cachedMessages = inputs
-		}
-	}
-
-	return inputs, nil
-}
-
 func (w *Worker) makeUnaryRequest(ctx *context.Context, reqMD *metadata.MD, input *dynamic.Message) error {
 	var res proto.Message
 	var resErr error
@@ -221,80 +182,127 @@ func (w *Worker) makeUnaryRequest(ctx *context.Context, reqMD *metadata.MD, inpu
 	return resErr
 }
 
-func (w *Worker) makeClientStreamingRequest(ctx *context.Context, input []*dynamic.Message) error {
+func (w *Worker) makeClientStreamingRequest(ctx *context.Context,
+	ctd *CallData, messageProvider StreamMessageProviderFunc) error {
 	var str *grpcdynamic.ClientStream
-	var err error
 	var callOptions = []grpc.CallOption{}
 	if w.config.enableCompression {
 		callOptions = append(callOptions, grpc.UseCompressor(gzip.Name))
 	}
-	str, err = w.stub.InvokeRpcClientStream(*ctx, w.mtd, callOptions...)
+	str, err := w.stub.InvokeRpcClientStream(*ctx, w.mtd, callOptions...)
+	if err != nil {
+		if w.config.hasLog {
+			w.config.log.Errorw("Invoke Client Streaming RPC call error: "+err.Error(), "workerID", w.workerID,
+				"call type", "client-streaming",
+				"call", w.mtd.GetFullyQualifiedName(), "error", err)
+		}
 
-	if err != nil && w.config.hasLog {
-		w.config.log.Errorw("Invoke Client Streaming RPC call error: "+err.Error(), "workerID", w.workerID,
-			"call type", "client-streaming",
-			"call", w.mtd.GetFullyQualifiedName(), "error", err)
+		return err
 	}
 
-	counter := 0
+	closeStream := func() {
+		res, closeErr := str.CloseAndReceive()
 
-	for err == nil {
-		inputLen := len(input)
-		if input == nil || inputLen == 0 {
-			res, closeErr := str.CloseAndReceive()
+		if w.config.hasLog {
+			w.config.log.Debugw("Close and receive", "workerID", w.workerID, "call type", "client-streaming",
+				"call", w.mtd.GetFullyQualifiedName(),
+				"response", res, "error", closeErr)
+		}
+	}
 
-			if w.config.hasLog {
-				w.config.log.Debugw("Close and receive", "workerID", w.workerID, "call type", "client-streaming",
-					"call", w.mtd.GetFullyQualifiedName(),
-					"response", res, "error", closeErr)
-			}
+	performSend := func(payload *dynamic.Message) (bool, error) {
+		err := str.SendMsg(payload)
 
-			break
+		if w.config.hasLog {
+			w.config.log.Debugw("Send message", "workerID", w.workerID, "call type", "client-streaming",
+				"call", w.mtd.GetFullyQualifiedName(),
+				"payload", payload, "error", err)
 		}
 
-		if counter == inputLen {
-			res, closeErr := str.CloseAndReceive()
+		if err == io.EOF {
+			return true, nil
+		}
 
-			if w.config.hasLog {
-				w.config.log.Debugw("Close and receive", "workerID", w.workerID, "call type", "client-streaming",
-					"call", w.mtd.GetFullyQualifiedName(),
-					"response", res, "error", closeErr)
+		return false, err
+	}
+
+	doneCh := make(chan struct{})
+	cancel := make(chan struct{}, 1)
+	if w.config.streamCallDuration > 0 {
+		go func() {
+			sct := time.NewTimer(w.config.streamCallDuration)
+			select {
+			case <-sct.C:
+				cancel <- struct{}{}
+				return
+			case <-doneCh:
+				if !sct.Stop() {
+					<-sct.C
+				}
+				return
 			}
+		}()
+	}
 
-			break
+	done := false
+	counter := uint(0)
+	end := false
+	for !done && len(cancel) == 0 {
+		// default message provider checks counter
+		// but we also need to keep our own counts
+		// in case of custom client providers
+
+		var payload *dynamic.Message
+		payload, err = messageProvider(ctd)
+
+		isLast := false
+		if errors.Is(err, ErrLastMessage) {
+			isLast = true
+			err = nil
 		}
 
-		payload := input[counter]
+		if err != nil {
+			if errors.Is(err, ErrEndStream) {
+				err = nil
+			}
+			break
+		}
 
-		var wait <-chan time.Time
-		if w.config.streamInterval > 0 {
-			wait = time.Tick(w.config.streamInterval)
-			<-wait
+		end, err = performSend(payload)
+		if end || err != nil || isLast || len(cancel) > 0 {
+			break
 		}
 
-		err = str.SendMsg(payload)
+		counter++
 
-		if w.config.hasLog {
-			w.config.log.Debugw("Send message", "workerID", w.workerID, "call type", "client-streaming",
-				"call", w.mtd.GetFullyQualifiedName(),
-				"payload", payload, "error", err)
+		if w.config.streamCallCount > 0 && counter >= w.config.streamCallCount {
+			break
 		}
 
-		if err == io.EOF {
-			// We get EOF on send if the server says "go away"
-			// We have to use CloseAndReceive to get the actual code
-			res, closeErr := str.CloseAndReceive()
-
-			if w.config.hasLog {
-				w.config.log.Debugw("Close and receive", "workerID", w.workerID, "call type", "client-streaming",
-					"call", w.mtd.GetFullyQualifiedName(),
-					"response", res, "error", closeErr)
+		if w.config.streamInterval > 0 {
+			wait := time.NewTimer(w.config.streamInterval)
+			select {
+			case <-wait.C:
+				break
+			case <-cancel:
+				if !wait.Stop() {
+					<-wait.C
+				}
+				done = true
+				break
 			}
-
-			break
 		}
-		counter++
 	}
+
+	for len(cancel) > 0 {
+		<-cancel
+	}
+
+	closeStream()
+
+	close(doneCh)
+	close(cancel)
+
 	return nil
 }
 
@@ -303,17 +311,53 @@ func (w *Worker) makeServerStreamingRequest(ctx *context.Context, input *dynamic
 	if w.config.enableCompression {
 		callOptions = append(callOptions, grpc.UseCompressor(gzip.Name))
 	}
-	str, err := w.stub.InvokeRpcServerStream(*ctx, w.mtd, input, callOptions...)
 
-	if err != nil && w.config.hasLog {
-		w.config.log.Errorw("Invoke Server Streaming RPC call error: "+err.Error(), "workerID", w.workerID,
-			"call type", "server-streaming",
-			"call", w.mtd.GetFullyQualifiedName(),
-			"input", input, "error", err)
+	callCtx, callCancel := context.WithCancel(*ctx)
+	defer callCancel()
+
+	str, err := w.stub.InvokeRpcServerStream(callCtx, w.mtd, input, callOptions...)
+
+	if err != nil {
+		if w.config.hasLog {
+			w.config.log.Errorw("Invoke Server Streaming RPC call error: "+err.Error(), "workerID", w.workerID,
+				"call type", "server-streaming",
+				"call", w.mtd.GetFullyQualifiedName(),
+				"input", input, "error", err)
+		}
+
+		return err
+	}
+
+	doneCh := make(chan struct{})
+	cancel := make(chan struct{}, 1)
+	if w.config.streamCallDuration > 0 {
+		go func() {
+			sct := time.NewTimer(w.config.streamCallDuration)
+			select {
+			case <-sct.C:
+				cancel <- struct{}{}
+				return
+			case <-doneCh:
+				if !sct.Stop() {
+					<-sct.C
+				}
+				return
+			}
+		}()
 	}
 
+	interceptCanceled := false
+	counter := uint(0)
 	for err == nil {
-		res, err := str.RecvMsg()
+		// we should check before receiving a message too
+		if w.config.streamCallDuration > 0 && len(cancel) > 0 {
+			<-cancel
+			callCancel()
+			break
+		}
+
+		var res proto.Message
+		res, err = str.RecvMsg()
 
 		if w.config.hasLog {
 			w.config.log.Debugw("Receive message", "workerID", w.workerID, "call type", "server-streaming",
@@ -321,25 +365,56 @@ func (w *Worker) makeServerStreamingRequest(ctx *context.Context, input *dynamic
 				"response", res, "error", err)
 		}
 
+		// 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
+
+		if w.streamRecv != nil {
+			if converted, ok := res.(*dynamic.Message); ok {
+				err = w.streamRecv(converted, err)
+				if errors.Is(err, ErrEndStream) && !interceptCanceled {
+					interceptCanceled = true
+					err = nil
+
+					callCancel()
+				}
+			}
+		}
+
 		if err != nil {
 			if err == io.EOF {
 				err = nil
 			}
+
 			break
 		}
+
+		counter++
+
+		if w.config.streamCallCount > 0 && counter >= w.config.streamCallCount {
+			callCancel()
+		}
+
+		if w.config.streamCallDuration > 0 && len(cancel) > 0 {
+			<-cancel
+			callCancel()
+		}
 	}
 
+	close(doneCh)
+	close(cancel)
+
 	return err
 }
 
-func (w *Worker) makeBidiRequest(ctx *context.Context, input []*dynamic.Message) error {
-	var str *grpcdynamic.BidiStream
-	var err error
+func (w *Worker) makeBidiRequest(ctx *context.Context,
+	ctd *CallData, messageProvider StreamMessageProviderFunc) error {
+
 	var callOptions = []grpc.CallOption{}
+
 	if w.config.enableCompression {
 		callOptions = append(callOptions, grpc.UseCompressor(gzip.Name))
 	}
-	str, err = w.stub.InvokeRpcBidiStream(*ctx, w.mtd, callOptions...)
+	str, err := w.stub.InvokeRpcBidiStream(*ctx, w.mtd, callOptions...)
 
 	if err != nil {
 		if w.config.hasLog {
@@ -351,73 +426,172 @@ func (w *Worker) makeBidiRequest(ctx *context.Context, input []*dynamic.Message)
 		return err
 	}
 
-	counter := 0
-
-	inputLen := len(input)
-
+	counter := uint(0)
+	indexCounter := 0
 	recvDone := make(chan bool)
+	sendDone := make(chan bool)
 
-	if input == nil || inputLen == 0 {
+	closeStream := func() {
 		closeErr := str.CloseSend()
 
 		if w.config.hasLog {
 			w.config.log.Debugw("Close send", "workerID", w.workerID, "call type", "bidi",
 				"call", w.mtd.GetFullyQualifiedName(), "error", closeErr)
 		}
+	}
 
-		return nil
+	doneCh := make(chan struct{})
+	cancel := make(chan struct{}, 1)
+	if w.config.streamCallDuration > 0 {
+		go func() {
+			sct := time.NewTimer(w.config.streamCallDuration)
+			select {
+			case <-sct.C:
+				cancel <- struct{}{}
+				return
+			case <-doneCh:
+				if !sct.Stop() {
+					<-sct.C
+				}
+				return
+			}
+		}()
 	}
 
+	var recvErr error
+
 	go func() {
-		for {
-			res, err := str.RecvMsg()
+		interceptCanceled := false
+
+		for recvErr == nil {
+			var res proto.Message
+			res, recvErr = str.RecvMsg()
 
 			if w.config.hasLog {
 				w.config.log.Debugw("Receive message", "workerID", w.workerID, "call type", "bidi",
 					"call", w.mtd.GetFullyQualifiedName(),
-					"response", res, "error", err)
+					"response", res, "error", recvErr)
 			}
 
-			if err != nil {
+			if w.streamRecv != nil {
+				if converted, ok := res.(*dynamic.Message); ok {
+					iErr := w.streamRecv(converted, recvErr)
+					if errors.Is(iErr, ErrEndStream) && !interceptCanceled {
+						interceptCanceled = true
+						if len(cancel) == 0 {
+							cancel <- struct{}{}
+						}
+						recvErr = nil
+					}
+				}
+			}
+
+			if recvErr != nil {
 				close(recvDone)
 				break
 			}
 		}
 	}()
 
-	for err == nil {
-		if counter == inputLen {
-			closeErr := str.CloseSend()
+	go func() {
+		done := false
+
+		for err == nil && !done {
+
+			// check at start before send too
+			if len(cancel) > 0 {
+				<-cancel
+				closeStream()
+				break
+			}
+
+			// default message provider checks counter
+			// but we also need to keep our own counts
+			// in case of custom client providers
+
+			var payload *dynamic.Message
+			payload, err = messageProvider(ctd)
+
+			isLast := false
+			if errors.Is(err, ErrLastMessage) {
+				isLast = true
+				err = nil
+			}
+
+			if err != nil {
+				if errors.Is(err, ErrEndStream) {
+					err = nil
+				}
+
+				closeStream()
+				break
+			}
+
+			err = str.SendMsg(payload)
+			if err != nil {
+				if err == io.EOF {
+					err = nil
+				}
+
+				break
+			}
 
 			if w.config.hasLog {
-				w.config.log.Debugw("Close send", "workerID", w.workerID, "call type", "bidi",
-					"call", w.mtd.GetFullyQualifiedName(), "error", closeErr)
+				w.config.log.Debugw("Send message", "workerID", w.workerID, "call type", "bidi",
+					"call", w.mtd.GetFullyQualifiedName(),
+					"payload", payload, "error", err)
 			}
 
-			break
-		}
+			if isLast {
+				closeStream()
+				break
+			}
 
-		payload := input[counter]
+			counter++
+			indexCounter++
 
-		var wait <-chan time.Time
-		if w.config.streamInterval > 0 {
-			wait = time.Tick(w.config.streamInterval)
-			<-wait
-		}
+			if w.config.streamCallCount > 0 && counter >= w.config.streamCallCount {
+				closeStream()
+				break
+			}
 
-		err = str.SendMsg(payload)
-		counter++
+			if len(cancel) > 0 {
+				<-cancel
+				closeStream()
+				break
+			}
 
-		if w.config.hasLog {
-			w.config.log.Debugw("Send message", "workerID", w.workerID, "call type", "bidi",
-				"call", w.mtd.GetFullyQualifiedName(),
-				"payload", payload, "error", err)
+			if w.config.streamInterval > 0 {
+				wait := time.NewTimer(w.config.streamInterval)
+				select {
+				case <-wait.C:
+					break
+				case <-cancel:
+					if !wait.Stop() {
+						<-wait.C
+					}
+					closeStream()
+					done = true
+					break
+				}
+			}
 		}
+
+		close(sendDone)
+	}()
+
+	_, _ = <-recvDone, <-sendDone
+
+	for len(cancel) > 0 {
+		<-cancel
 	}
 
-	if err == nil {
-		<-recvDone
+	close(doneCh)
+	close(cancel)
+
+	if err == nil && recvErr != nil {
+		err = recvErr
 	}
 
-	return nil
+	return err
 }
diff --git a/www/docs/intro.md b/www/docs/intro.md
index 3a315c4..534eb51 100644
--- a/www/docs/intro.md
+++ b/www/docs/intro.md
@@ -10,6 +10,6 @@ title: Introduction
 [![Donate](https://img.shields.io/badge/Donate-PayPal-green.svg?style=flat-square)](https://www.paypal.me/bojandj)
 [![Buy me a coffee](https://img.shields.io/badge/buy%20me-a%20coffee-orange.svg?style=flat-square)](https://www.buymeacoffee.com/bojand)
 
-`ghz` is a simple command line utility and [Go](http://golang.org/) package for load testing and benchmarking [gRPC](http://grpc.io) services. It is intended to be used for testing and debugging services locally, and in automated continous intergration environments for performance regression testing.
+`ghz` is a command line utility and [Go](http://golang.org/) package for load testing and benchmarking [gRPC](http://grpc.io) services. It is intended to be used for testing and debugging services locally, and in automated continous intergration environments for performance regression testing.
 
 Additionally the core of the command line tool is implemented as a Go library package that can be used to programatically implement performance tests as well.
diff --git a/www/docs/options.md b/www/docs/options.md
index 76876cd..47d763c 100644
--- a/www/docs/options.md
+++ b/www/docs/options.md
@@ -243,7 +243,67 @@ Path for call metadata JSON file. For example, `-M /home/user/metadata.json` or
 
 ### `--stream-interval`
 
-Stream interval duration. Spread stream sends by given amount. Only applies to client and bidi streaming calls. Example: `100ms`
+Stream interval duration. Spread stream sends by given amount. Only applies to client and bidi streaming calls. Example: `100ms`.
+
+### `--stream-call-duration`
+
+Maximum stream call duration. For client streaming and bidi calls, we'll send messages until this duration expires.
+
+For server streaming calls we will receive message until the duration has expired. Note that in server streaming calls the cancellation will result in call cancelled error.  
+
+Example: `500ms`.
+
+### `--stream-call-count`
+
+The maximum number of message sends or receives the client will perform in a streaming call before closing the stream and ending the call. For client and bidi streaming calls this dictates the number of messages we will send.
+
+If the data array contains more elements than the count, only data up to the number specified will be used.
+
+If the data array contains fewer elements than the count specified, all the data will be iterated over repeatedly until count limit is reached. 
+
+For server streaming calls we will receive message until the specified count is reached. Note that in server streaming calls the cancellation will result in call cancelled error.  
+
+Examples:
+
+```sh
+--stream-call-count=2 -d '[{"name":"Joe"},{"name":"Kate"},{"name":"Sara"}]'
+```
+
+Will cause only `[{"name":"Joe"},{"name":"Kate"}]` to be sent. Similarly:
+
+```sh
+--stream-call-count=5 -d '[{"name":"Joe"},{"name":"Kate"},{"name":"Sara"}]'
+```
+
+Will cause `[{"name":"Joe"},{"name":"Kate"},{"name":"Sara"},{"name":"Joe"},{"name":"Kate"}]` to be sent.
+
+### `--stream-dynamic-messages`
+
+In streaming calls, regenerate and apply call template data on every message send operation.
+This is helpful in combination with template functionality to generate data _for every message sent_ in a streaming call.
+For example:
+
+```sh
+--stream-dynamic-messages=true --stream-call-count=5 -d '{"name":"{{randomString 8 }}"}'
+```
+
+Will result in streaming call with the following data sent:
+
+```json
+[{"name":"sKNdMCIb"},{"name":"KLVXDvn1"},{"name":"RJ3knnBh"},{"name":"FTBqQ7nl"},{"name":"FzeMQIWo"}]
+```
+
+Contrast that with the default dynamic messages setting turned off; which means the template data will be applied only once for each stream call request, but _not_ for each message sent _in_ the streaming call.
+
+```sh
+--stream-call-count=5 -d '{"name":"{{randomString 8 }}"}'
+```
+
+Results in the following data sent:
+
+```json
+[{"name":"5hL64dd0"},{"name":"5hL64dd0"},{"name":"5hL64dd0"},{"name":"5hL64dd0"},{"name":"5hL64dd0"}]
+```
 
 ### `--reflect-metadata`
 
@@ -313,6 +373,10 @@ ghz --insecure \
 
 Enable gzip compression on requests.
 
+### `--count-errors`
+
+By default stats for fastest, slowest, average, histogram, and latency distributions only take into account the responses with OK status. This option enabled counting of erroneous (non-OK) responses in stats calculations as well.
+
 ### `-v`, `--version`
 
 Print the version.
diff --git a/www/docs/usage.md b/www/docs/usage.md
index a8af9be..e8b31a6 100644
--- a/www/docs/usage.md
+++ b/www/docs/usage.md
@@ -7,62 +7,66 @@ title: Usage
 usage: ghz [<flags>] [<host>]
 
 Flags:
-  -h, --help                   Show context-sensitive help (also try --help-long and --help-man).
-      --config=                Path to the JSON or TOML config file that specifies all the test run settings.
-      --proto=                 The Protocol Buffer .proto file.
-      --protoset=              The compiled protoset file. Alternative to proto. -proto takes precedence.
-      --call=                  A fully-qualified method name in 'package.Service/method' or 'package.Service.Method' format.
-  -i, --import-paths=          Comma separated list of proto import paths. The current working directory and the directory of the protocol buffer file are automatically added to the import list.
-      --cacert=                File containing trusted root certificates for verifying the server.
-      --cert=                  File containing client certificate (public key), to present to the server. Must also provide -key option.
-      --key=                   File containing client private key, to present to the server. Must also provide -cert option.
-      --cname=                 Server name override when validating TLS certificate - useful for self signed certs.
-      --skipTLS                Skip TLS client verification of the server's certificate chain and host name.
-      --insecure               Use plaintext and insecure connection.
-      --authority=             Value to be used as the :authority pseudo-header. Only works if -insecure is used.
-      --async                  Make requests asynchronous as soon as possible. Does not wait for request to finish before sending next one.
-  -r, --rps=0                  Requests per second (RPS) rate limit for constant load schedule. Default is no rate limit.
-      --load-schedule="const"  Specifies the load schedule. Options are const, step, or line. Default is const.
-      --load-start=0           Specifies the RPS load start value for step or line schedules.
-      --load-step=0            Specifies the load step value or slope value.
-      --load-end=0             Specifies the load end value for step or line load schedules.
-      --load-step-duration=0   Specifies the load step duration value for step load schedule.
-      --load-max-duration=0    Specifies the max load duration value for step or line load schedule.
-  -c, --concurrency=50         Number of request workers to run concurrently for const concurrency schedule. Default is 50.
+  -h, --help                     Show context-sensitive help (also try --help-long and --help-man).
+      --config=                  Path to the JSON or TOML config file that specifies all the test run settings.
+      --proto=                   The Protocol Buffer .proto file.
+      --protoset=                The compiled protoset file. Alternative to proto. -proto takes precedence.
+      --call=                    A fully-qualified method name in 'package.Service/method' or 'package.Service.Method' format.
+  -i, --import-paths=            Comma separated list of proto import paths. The current working directory and the directory of the protocol buffer file are automatically added to the import list.
+      --cacert=                  File containing trusted root certificates for verifying the server.
+      --cert=                    File containing client certificate (public key), to present to the server. Must also provide -key option.
+      --key=                     File containing client private key, to present to the server. Must also provide -cert option.
+      --cname=                   Server name override when validating TLS certificate - useful for self signed certs.
+      --skipTLS                  Skip TLS client verification of the server's certificate chain and host name.
+      --insecure                 Use plaintext and insecure connection.
+      --authority=               Value to be used as the :authority pseudo-header. Only works if -insecure is used.
+      --async                    Make requests asynchronous as soon as possible. Does not wait for request to finish before sending next one.
+  -r, --rps=0                    Requests per second (RPS) rate limit for constant load schedule. Default is no rate limit.
+      --load-schedule="const"    Specifies the load schedule. Options are const, step, or line. Default is const.
+      --load-start=0             Specifies the RPS load start value for step or line schedules.
+      --load-step=0              Specifies the load step value or slope value.
+      --load-end=0               Specifies the load end value for step or line load schedules.
+      --load-step-duration=0     Specifies the load step duration value for step load schedule.
+      --load-max-duration=0      Specifies the max load duration value for step or line load schedule.
+  -c, --concurrency=50           Number of request workers to run concurrently for const concurrency schedule. Default is 50.
       --concurrency-schedule="const"
-                               Concurrency change schedule. Options are const, step, or line. Default is const.
-      --concurrency-start=0    Concurrency start value for step and line concurrency schedules.
-      --concurrency-end=0      Concurrency end value for step and line concurrency schedules.
-      --concurrency-step=1     Concurrency step / slope value for step and line concurrency schedules.
+                                 Concurrency change schedule. Options are const, step, or line. Default is const.
+      --concurrency-start=0      Concurrency start value for step and line concurrency schedules.
+      --concurrency-end=0        Concurrency end value for step and line concurrency schedules.
+      --concurrency-step=1       Concurrency step / slope value for step and line concurrency schedules.
       --concurrency-step-duration=0
-                               Specifies the concurrency step duration value for step concurrency schedule.
+                                 Specifies the concurrency step duration value for step concurrency schedule.
       --concurrency-max-duration=0
-                               Specifies the max concurrency adjustment duration value for step or line concurrency schedule.
-  -n, --total=200              Number of requests to run. Default is 200.
-  -t, --timeout=20s            Timeout for each request. Default is 20s, use 0 for infinite.
-  -z, --duration=0             Duration of application to send requests. When duration is reached, application stops and exits. If duration is specified, n is ignored. Examples: -z 10s -z 3m.
-  -x, --max-duration=0         Maximum duration of application to send requests with n setting respected. If duration is reached before n requests are completed, application stops and exits. Examples: -x 10s -x 3m.
-      --duration-stop="close"  Specifies how duration stop is reported. Options are close, wait or ignore. Default is close.
-  -d, --data=                  The call data as stringified JSON. If the value is '@' then the request contents are read from stdin.
-  -D, --data-file=             File path for call data JSON file. Examples: /home/user/file.json or ./file.json.
-  -b, --binary                 The call data comes as serialized binary message or multiple count-prefixed messages read from stdin.
-  -B, --binary-file=           File path for the call data as serialized binary message or multiple count-prefixed messages.
-  -m, --metadata=              Request metadata as stringified JSON.
-  -M, --metadata-file=         File path for call metadata JSON file. Examples: /home/user/metadata.json or ./metadata.json.
-      --stream-interval=0      Interval for stream requests between message sends.
-      --reflect-metadata=      Reflect metadata as stringified JSON used only for reflection request.
-  -o, --output=                Output path. If none provided stdout is used.
-  -O, --format=                Output format. One of: summary, csv, json, pretty, html, influx-summary, influx-details. Default is summary.
-      --skipFirst=0            Skip the first X requests when doing the results tally.
-      --connections=1          Number of connections to use. Concurrency is distributed evenly among all the connections. Default is 1.
-      --connect-timeout=10s    Connection timeout for the initial connection dial. Default is 10s.
-      --keepalive=0            Keepalive time duration. Only used if present and above 0.
-      --name=                  User specified name for the test.
-      --tags=                  JSON representation of user-defined string tags.
-      --cpus=12                Number of cpu cores to use.
-      --debug=                 The path to debug log file.
-  -e, --enable-compression     Enable Gzip compression on requests.
-  -v, --version                Show application version.
+                                 Specifies the max concurrency adjustment duration value for step or line concurrency schedule.
+  -n, --total=200                Number of requests to run. Default is 200.
+  -t, --timeout=20s              Timeout for each request. Default is 20s, use 0 for infinite.
+  -z, --duration=0               Duration of application to send requests. When duration is reached, application stops and exits. If duration is specified, n is ignored. Examples: -z 10s -z 3m.
+  -x, --max-duration=0           Maximum duration of application to send requests with n setting respected. If duration is reached before n requests are completed, application stops and exits. Examples: -x 10s -x 3m.
+      --duration-stop="close"    Specifies how duration stop is reported. Options are close, wait or ignore. Default is close.
+  -d, --data=                    The call data as stringified JSON. If the value is '@' then the request contents are read from stdin.
+  -D, --data-file=               File path for call data JSON file. Examples: /home/user/file.json or ./file.json.
+  -b, --binary                   The call data comes as serialized binary message or multiple count-prefixed messages read from stdin.
+  -B, --binary-file=             File path for the call data as serialized binary message or multiple count-prefixed messages.
+  -m, --metadata=                Request metadata as stringified JSON.
+  -M, --metadata-file=           File path for call metadata JSON file. Examples: /home/user/metadata.json or ./metadata.json.
+      --stream-interval=0        Interval for stream requests between message sends.
+      --stream-call-duration=0   Duration after which client will close the stream in each streaming call.
+      --stream-call-count=0      Count of messages sent, after which client will close the stream in each streaming call.
+      --stream-dynamic-messages  In streaming calls, regenerate and apply call template data on every message send.
+      --reflect-metadata=        Reflect metadata as stringified JSON used only for reflection request.
+  -o, --output=                  Output path. If none provided stdout is used.
+  -O, --format=                  Output format. One of: summary, csv, json, pretty, html, influx-summary, influx-details. Default is summary.
+      --skipFirst=0              Skip the first X requests when doing the results tally.
+      --count-errors             Count erroneous (non-OK) resoponses in stats calculations.
+      --connections=1            Number of connections to use. Concurrency is distributed evenly among all the connections. Default is 1.
+      --connect-timeout=10s      Connection timeout for the initial connection dial. Default is 10s.
+      --keepalive=0              Keepalive time duration. Only used if present and above 0.
+      --name=                    User specified name for the test.
+      --tags=                    JSON representation of user-defined string tags.
+      --cpus=12                  Number of cpu cores to use.
+      --debug=                   The path to debug log file.
+  -e, --enable-compression       Enable Gzip compression on requests.
+  -v, --version                  Show application version.
 
 Args:
   [<host>]  Host and port to test.
diff --git a/www/website/i18n/en.json b/www/website/i18n/en.json
index 9fdccf2..e57fbec 100644
--- a/www/website/i18n/en.json
+++ b/www/website/i18n/en.json
@@ -3,7 +3,7 @@
   "localized-strings": {
     "next": "Next",
     "previous": "Previous",
-    "tagline": "Simple gRPC benchmarking and load testing tool ",
+    "tagline": "gRPC benchmarking and load testing tool",
     "docs": {
       "calldata": {
         "title": "Call Template Data"
diff --git a/www/website/siteConfig.js b/www/website/siteConfig.js
index e58506e..fed7b04 100755
--- a/www/website/siteConfig.js
+++ b/www/website/siteConfig.js
@@ -5,7 +5,7 @@ const users = [];
 
 const siteConfig = {
   title: 'ghz', // Title for your website.
-  tagline: 'Simple gRPC benchmarking and load testing tool ',
+  tagline: 'gRPC benchmarking and load testing tool',
   url: 'https://ghz.sh', // Your website URL
   baseUrl: '/', // Base URL for your project */
 


Test patch:
diff --git a/load/pacer_test.go b/load/pacer_test.go
index d9350ab..8ea16e5 100644
--- a/load/pacer_test.go
+++ b/load/pacer_test.go
@@ -10,8 +10,6 @@ import (
 )
 
 func TestConstantPacer(t *testing.T) {
-	t.Parallel()
-
 	for _, tc := range []struct {
 		freq    uint64
 		max     uint64
@@ -135,8 +133,6 @@ func TestConstantPacer(t *testing.T) {
 }
 
 func TestConstantPacer_Rate(t *testing.T) {
-	t.Parallel()
-
 	for _, tc := range []struct {
 		freq    uint64
 		elapsed time.Duration
@@ -167,8 +163,6 @@ func TestConstantPacer_String(t *testing.T) {
 
 func TestLinearPacer(t *testing.T) {
 
-	t.Parallel()
-
 	for ti, tt := range []struct {
 		// pacer config
 		start        uint64
@@ -422,7 +416,6 @@ func TestLinearPacer(t *testing.T) {
 }
 
 func TestStepPacer_hits(t *testing.T) {
-	t.Parallel()
 
 	// TODO improve this to have different pacer params
 	p := StepPacer{
@@ -448,8 +441,6 @@ func TestStepPacer_hits(t *testing.T) {
 }
 
 func TestStepPacer_Rate(t *testing.T) {
-	t.Parallel()
-
 	for _, tc := range []struct {
 		// pacer config
 		start        uint64
@@ -769,7 +760,6 @@ func TestStepPacer_Rate(t *testing.T) {
 }
 
 func TestStepPacer(t *testing.T) {
-	t.Parallel()
 
 	for ti, tc := range []struct {
 		// pacer config
diff --git a/load/worker_ticker_test.go b/load/worker_ticker_test.go
index fe61e79..1cdc64f 100644
--- a/load/worker_ticker_test.go
+++ b/load/worker_ticker_test.go
@@ -26,8 +26,6 @@ func TestConstWorkerTicker(t *testing.T) {
 }
 
 func TestStepWorkerTicker(t *testing.T) {
-	t.Parallel()
-
 	t.Run("step increase load duration", func(t *testing.T) {
 		wt := StepWorkerTicker{
 			C:            make(chan TickValue),
diff --git a/runner/calldata_test.go b/runner/calldata_test.go
index db416c3..f104883 100644
--- a/runner/calldata_test.go
+++ b/runner/calldata_test.go
@@ -11,6 +11,7 @@ import (
 )
 
 func TestCallData_New(t *testing.T) {
+
 	md, err := protodesc.GetMethodDescFromProto("helloworld.Greeter/SayHello", "../testdata/greeter.proto", []string{})
 	assert.NoError(t, err)
 	assert.NotNil(t, md)
@@ -41,10 +42,6 @@ func TestCallData_ExecuteData(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, md)
 
-	ctd := newCallData(md, nil, "worker_id_123", 200)
-
-	assert.NotNil(t, ctd)
-
 	var tests = []struct {
 		name        string
 		in          string
@@ -75,7 +72,10 @@ func TestCallData_ExecuteData(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			r, err := ctd.executeData(tt.in)
+			ctd := newCallData(md, nil, "worker_id_123", 200)
+			assert.NotNil(t, ctd)
+
+			r, err := ctd.ExecuteData(tt.in)
 			if tt.expectError {
 				assert.Error(t, err)
 			} else {
@@ -92,10 +92,6 @@ func TestCallData_ExecuteMetadata(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, md)
 
-	ctd := newCallData(md, nil, "worker_id_123", 200)
-
-	assert.NotNil(t, ctd)
-
 	var tests = []struct {
 		name        string
 		in          string
@@ -121,6 +117,10 @@ func TestCallData_ExecuteMetadata(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+
+			ctd := newCallData(md, nil, "worker_id_123", 200)
+			assert.NotNil(t, ctd)
+
 			r, err := ctd.executeMetadata(tt.in)
 			if tt.expectError {
 				assert.Error(t, err)
@@ -138,14 +138,13 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, md)
 
-	ctd := newCallData(md, nil, "worker_id_123", 200)
-
-	assert.NotNil(t, ctd)
-
 	t.Run("newUUID", func(t *testing.T) {
 
+		ctd := newCallData(md, nil, "worker_id_123", 200)
+		assert.NotNil(t, ctd)
+
 		// no template
-		r, err := ctd.executeData(`{"trace_id":"asdf"}`)
+		r, err := ctd.ExecuteData(`{"trace_id":"asdf"}`)
 		assert.NoError(t, err)
 		assert.Equal(t, `{"trace_id":"asdf"}`, string(r))
 
@@ -154,7 +153,7 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 		assert.Equal(t, map[string]string{"trace_id": "asdf"}, rm)
 
 		// new uuid
-		r, err = ctd.executeData(`{"trace_id":"{{newUUID}}"}`)
+		r, err = ctd.ExecuteData(`{"trace_id":"{{newUUID}}"}`)
 		assert.NoError(t, err)
 		rs := strings.Replace(string(r), `{"trace_id":"`, "", -1)
 		rs = strings.Replace(rs, `"}`, "", -1)
@@ -189,9 +188,11 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 	})
 
 	t.Run("randomString", func(t *testing.T) {
+		ctd := newCallData(md, nil, "worker_id_123", 200)
+		assert.NotNil(t, ctd)
 
 		// no template
-		r, err := ctd.executeData(`{"trace_id":"asdf"}`)
+		r, err := ctd.ExecuteData(`{"trace_id":"asdf"}`)
 		assert.NoError(t, err)
 		assert.Equal(t, `{"trace_id":"asdf"}`, string(r))
 
@@ -200,7 +201,7 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 		assert.Equal(t, map[string]string{"trace_id": "asdf"}, rm)
 
 		// default length when 0
-		r, err = ctd.executeData(`{"trace_id":"{{randomString 0}}"}`)
+		r, err = ctd.ExecuteData(`{"trace_id":"{{randomString 0}}"}`)
 		assert.NoError(t, err)
 		rs := strings.Replace(string(r), `{"trace_id":"`, "", -1)
 		rs = strings.Replace(rs, `"}`, "", -1)
@@ -209,7 +210,7 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 		assert.True(t, len(rs) <= 16)
 
 		// default length when -1
-		r, err = ctd.executeData(`{"trace_id":"{{randomString -1}}"}`)
+		r, err = ctd.ExecuteData(`{"trace_id":"{{randomString -1}}"}`)
 		assert.NoError(t, err)
 		rs2 := strings.Replace(string(r), `{"trace_id":"`, "", -1)
 		rs2 = strings.Replace(rs2, `"}`, "", -1)
@@ -219,7 +220,7 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 		assert.True(t, len(rs2) <= 16)
 
 		// specific length
-		r, err = ctd.executeData(`{"trace_id":"{{randomString 10}}"}`)
+		r, err = ctd.ExecuteData(`{"trace_id":"{{randomString 10}}"}`)
 		assert.NoError(t, err)
 		rs = strings.Replace(string(r), `{"trace_id":"`, "", -1)
 		rs = strings.Replace(rs, `"}`, "", -1)
@@ -249,6 +250,10 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 	})
 
 	t.Run("custom functions", func(t *testing.T) {
+
+		ctd := newCallData(md, nil, "worker_id_123", 200)
+		assert.NotNil(t, ctd)
+
 		ctd = newCallData(md, template.FuncMap{
 			"getSKU": func() string {
 				return "custom-sku"
@@ -258,7 +263,7 @@ func TestCallTemplateData_ExecuteFuncs(t *testing.T) {
 			},
 		}, "worker_id_123", 200)
 
-		r, err := ctd.executeData(`{"trace_id":"{{newUUID}}", "span_id":"{{getSKU}}"}`)
+		r, err := ctd.ExecuteData(`{"trace_id":"{{newUUID}}", "span_id":"{{getSKU}}"}`)
 		assert.NoError(t, err)
 		assert.Equal(t, `{"trace_id":"custom-uuid", "span_id":"custom-sku"}`, string(r))
 
diff --git a/runner/config_test.go b/runner/config_test.go
index e82eb57..5a76c91 100644
--- a/runner/config_test.go
+++ b/runner/config_test.go
@@ -1,6 +1,7 @@
 package runner
 
 import (
+	"encoding/json"
 	"strconv"
 	"testing"
 	"time"
@@ -106,3 +107,17 @@ func TestConfig_Load(t *testing.T) {
 		})
 	}
 }
+
+func TestConfig_MarshalJSON(t *testing.T) {
+	cfg := Config{
+		Insecure: true,
+		Proto:    "proto/service.proto",
+		Call:     "grpcbin.GRPCBin.DummyUnary",
+		Data:     "{interval_in_seconds:10,latitude:-23.43,longitude:-46.45,radius:5000}",
+		Host:     "localhost:8080",
+	}
+
+	_, err := json.Marshal(cfg)
+
+	assert.NoError(t, err)
+}
diff --git a/runner/options_test.go b/runner/options_test.go
index d8501b3..e23a945 100644
--- a/runner/options_test.go
+++ b/runner/options_test.go
@@ -307,7 +307,6 @@ func TestRunConfig_newRunConfig(t *testing.T) {
 	})
 
 	t.Run("with data from reader", func(t *testing.T) {
-
 		file, _ := os.Open("../testdata/data.json")
 		defer file.Close()
 
@@ -343,7 +342,6 @@ func TestRunConfig_newRunConfig(t *testing.T) {
 	})
 
 	t.Run("with connections", func(t *testing.T) {
-
 		file, _ := os.Open("../testdata/data.json")
 		defer file.Close()
 
@@ -380,7 +378,6 @@ func TestRunConfig_newRunConfig(t *testing.T) {
 	})
 
 	t.Run("with invalid connections > concurrency", func(t *testing.T) {
-
 		file, _ := os.Open("../testdata/data.json")
 		defer file.Close()
 
diff --git a/runner/run_test.go b/runner/run_test.go
index e376f71..ce59de0 100644
--- a/runner/run_test.go
+++ b/runner/run_test.go
@@ -9,8 +9,11 @@ import (
 	"github.com/bojand/ghz/internal"
 	"github.com/bojand/ghz/internal/helloworld"
 	"github.com/golang/protobuf/proto"
+	"github.com/google/uuid"
 	"github.com/jhump/protoreflect/desc"
+	"github.com/jhump/protoreflect/dynamic"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/metadata"
 )
 
 func changeFunc(mtd *desc.MethodDescriptor, cd *CallData) []byte {
@@ -563,133 +566,149 @@ func TestRunUnary(t *testing.T) {
 
 		assert.Equal(t, []string{"0", "1", "2", "0", "1", "2"}, names)
 	})
-}
-
-func TestRunServerStreaming(t *testing.T) {
-	callType := helloworld.ServerStream
-
-	gs, s, err := internal.StartServer(false)
-
-	if err != nil {
-		assert.FailNow(t, err.Error())
-	}
-
-	defer s.Stop()
-
-	gs.ResetCounters()
-
-	data := make(map[string]interface{})
-	data["name"] = "bob"
-
-	report, err := Run(
-		"helloworld.Greeter.SayHellos",
-		internal.TestLocalhost,
-		WithProtoFile("../testdata/greeter.proto", []string{}),
-		WithTotalRequests(15),
-		WithConcurrency(3),
-		WithTimeout(time.Duration(20*time.Second)),
-		WithDialTimeout(time.Duration(20*time.Second)),
-		WithData(data),
-		WithInsecure(true),
-		WithName("server streaming test"),
-	)
 
-	assert.NoError(t, err)
+	t.Run("with data provider", func(t *testing.T) {
+		gs.ResetCounters()
 
-	assert.NotNil(t, report)
+		callCounter := 0
 
-	assert.Equal(t, 15, int(report.Count))
-	assert.NotZero(t, report.Average)
-	assert.NotZero(t, report.Fastest)
-	assert.NotZero(t, report.Slowest)
-	assert.NotZero(t, report.Rps)
-	assert.Equal(t, "server streaming test", report.Name)
-	assert.NotEmpty(t, report.Date)
-	assert.NotEmpty(t, report.Options)
-	assert.NotEmpty(t, report.Details)
-	assert.Equal(t, true, report.Options.Insecure)
-	assert.NotEmpty(t, report.LatencyDistribution)
-	assert.Equal(t, ReasonNormalEnd, report.EndReason)
-	assert.Empty(t, report.ErrorDist)
+		report, err := Run(
+			"helloworld.Greeter.SayHello",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(5),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithDataProvider(func(*CallData) ([]*dynamic.Message, error) {
+				protoMsg := &helloworld.HelloRequest{Name: strconv.Itoa(callCounter)}
+				dynamicMsg, err := dynamic.AsDynamicMessage(protoMsg)
+				if err != nil {
+					return nil, err
+				}
 
-	assert.NotEqual(t, report.Average, report.Slowest)
-	assert.NotEqual(t, report.Average, report.Fastest)
-	assert.NotEqual(t, report.Slowest, report.Fastest)
+				callCounter++
 
-	count := gs.GetCount(callType)
-	assert.Equal(t, 15, count)
+				return []*dynamic.Message{dynamicMsg}, nil
+			}),
+			WithInsecure(true),
+		)
 
-	connCount := gs.GetConnectionCount()
-	assert.Equal(t, 1, connCount)
-}
+		assert.NoError(t, err)
 
-func TestRunClientStreaming(t *testing.T) {
-	callType := helloworld.ClientStream
+		assert.NotNil(t, report)
 
-	gs, s, err := internal.StartServer(false)
+		assert.Equal(t, 5, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.Details)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
 
-	if err != nil {
-		assert.FailNow(t, err.Error())
-	}
+		assert.NotEqual(t, report.Average, report.Slowest)
+		assert.NotEqual(t, report.Average, report.Fastest)
+		assert.NotEqual(t, report.Slowest, report.Fastest)
 
-	defer s.Stop()
+		count := gs.GetCount(callType)
+		assert.Equal(t, 5, count)
 
-	gs.ResetCounters()
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 5)
+		names := make([]string, 0)
+		for _, msgs := range calls {
+			for _, msg := range msgs {
+				names = append(names, msg.GetName())
+			}
+		}
 
-	m1 := make(map[string]interface{})
-	m1["name"] = "bob"
+		assert.Equal(t, []string{"0", "1", "2", "3", "4"}, names)
+	})
 
-	m2 := make(map[string]interface{})
-	m2["name"] = "Kate"
+	t.Run("with metadata provider", func(t *testing.T) {
+		gs.ResetCounters()
 
-	m3 := make(map[string]interface{})
-	m3["name"] = "foo"
+		callCounter := 0
 
-	data := []interface{}{m1, m2, m3}
+		report, err := Run(
+			"helloworld.Greeter.SayHello",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(5),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithDataProvider(func(*CallData) ([]*dynamic.Message, error) {
+				name := strconv.Itoa(callCounter)
+				if callCounter == 1 || callCounter == 3 {
+					name = "__record_metadata__"
+				}
+				protoMsg := &helloworld.HelloRequest{Name: name}
+				dynamicMsg, err := dynamic.AsDynamicMessage(protoMsg)
+				if err != nil {
+					return nil, err
+				}
+
+				callCounter++
+
+				return []*dynamic.Message{dynamicMsg}, nil
+			}),
+			WithMetadataProvider(func(*CallData) (*metadata.MD, error) {
+				mdv := "secret" + strconv.Itoa(callCounter)
+				return &metadata.MD{"token": []string{mdv}}, nil
+			}),
+			WithInsecure(true),
+		)
 
-	report, err := Run(
-		"helloworld.Greeter.SayHelloCS",
-		internal.TestLocalhost,
-		WithProtoFile("../testdata/greeter.proto", []string{}),
-		WithTotalRequests(16),
-		WithConcurrency(4),
-		WithTimeout(time.Duration(20*time.Second)),
-		WithDialTimeout(time.Duration(20*time.Second)),
-		WithData(data),
-		WithInsecure(true),
-	)
+		assert.NoError(t, err)
 
-	assert.NoError(t, err)
+		assert.NotNil(t, report)
 
-	assert.NotNil(t, report)
+		assert.Equal(t, 5, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.Details)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
 
-	assert.Equal(t, 16, int(report.Count))
-	assert.NotZero(t, report.Average)
-	assert.NotZero(t, report.Fastest)
-	assert.NotZero(t, report.Slowest)
-	assert.NotZero(t, report.Rps)
-	assert.Empty(t, report.Name)
-	assert.NotEmpty(t, report.Date)
-	assert.NotEmpty(t, report.Details)
-	assert.NotEmpty(t, report.Options)
-	assert.Equal(t, true, report.Options.Insecure)
-	assert.NotEmpty(t, report.LatencyDistribution)
-	assert.Equal(t, ReasonNormalEnd, report.EndReason)
-	assert.Empty(t, report.ErrorDist)
+		assert.NotEqual(t, report.Average, report.Slowest)
+		assert.NotEqual(t, report.Average, report.Fastest)
+		assert.NotEqual(t, report.Slowest, report.Fastest)
 
-	assert.NotEqual(t, report.Average, report.Slowest)
-	assert.NotEqual(t, report.Average, report.Fastest)
-	assert.NotEqual(t, report.Slowest, report.Fastest)
+		count := gs.GetCount(callType)
+		assert.Equal(t, 5, count)
 
-	count := gs.GetCount(callType)
-	assert.Equal(t, 16, count)
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 5)
+		names := make([]string, 0)
+		for _, msgs := range calls {
+			for _, msg := range msgs {
+				names = append(names, msg.GetName())
+			}
+		}
 
-	connCount := gs.GetConnectionCount()
-	assert.Equal(t, 1, connCount)
+		assert.Equal(t, []string{"0", "__record_metadata__||token:secret2", "2", "__record_metadata__||token:secret4", "4"}, names)
+	})
 }
 
-func TestRunClientStreamingBinary(t *testing.T) {
-	callType := helloworld.ClientStream
+func TestRunServerStreaming(t *testing.T) {
+
+	callType := helloworld.ServerStream
 
 	gs, s, err := internal.StartServer(false)
 
@@ -699,121 +718,1568 @@ func TestRunClientStreamingBinary(t *testing.T) {
 
 	defer s.Stop()
 
-	gs.ResetCounters()
-
-	msg := &helloworld.HelloRequest{}
-	msg.Name = "bob"
-
-	binData, err := proto.Marshal(msg)
-	assert.NoError(t, err)
-
-	report, err := Run(
-		"helloworld.Greeter.SayHelloCS",
-		internal.TestLocalhost,
-		WithProtoFile("../testdata/greeter.proto", []string{}),
-		WithTotalRequests(24),
-		WithConcurrency(4),
-		WithTimeout(time.Duration(20*time.Second)),
-		WithDialTimeout(time.Duration(20*time.Second)),
-		WithBinaryData(binData),
-		WithInsecure(true),
-	)
+	t.Run("basic", func(t *testing.T) {
 
-	assert.NoError(t, err)
+		gs.ResetCounters()
 
-	assert.NotNil(t, report)
+		data := make(map[string]interface{})
+		data["name"] = "bob"
 
-	assert.Equal(t, 24, int(report.Count))
-	assert.NotZero(t, report.Average)
-	assert.NotZero(t, report.Fastest)
-	assert.NotZero(t, report.Slowest)
-	assert.NotZero(t, report.Rps)
-	assert.Empty(t, report.Name)
-	assert.NotEmpty(t, report.Date)
-	assert.NotEmpty(t, report.Details)
-	assert.NotEmpty(t, report.Options)
-	assert.Equal(t, true, report.Options.Insecure)
-	assert.NotEmpty(t, report.LatencyDistribution)
-	assert.Equal(t, ReasonNormalEnd, report.EndReason)
-	assert.Empty(t, report.ErrorDist)
+		report, err := Run(
+			"helloworld.Greeter.SayHellos",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(15),
+			WithConcurrency(3),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithData(data),
+			WithInsecure(true),
+			WithName("server streaming test"),
+		)
 
-	assert.NotEqual(t, report.Average, report.Slowest)
-	assert.NotEqual(t, report.Average, report.Fastest)
-	assert.NotEqual(t, report.Slowest, report.Fastest)
+		assert.NoError(t, err)
 
-	count := gs.GetCount(callType)
-	assert.Equal(t, 24, count)
+		assert.NotNil(t, report)
 
-	connCount := gs.GetConnectionCount()
-	assert.Equal(t, 1, connCount)
-}
+		assert.Equal(t, 15, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Equal(t, "server streaming test", report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.Details)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
 
-func TestRunBidi(t *testing.T) {
-	callType := helloworld.Bidi
+		assert.NotEqual(t, report.Average, report.Slowest)
+		assert.NotEqual(t, report.Average, report.Fastest)
+		assert.NotEqual(t, report.Slowest, report.Fastest)
 
-	gs, s, err := internal.StartServer(false)
+		count := gs.GetCount(callType)
+		assert.Equal(t, 15, count)
 
-	if err != nil {
-		assert.FailNow(t, err.Error())
-	}
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+	})
 
-	defer s.Stop()
+	t.Run("with stream cancel", func(t *testing.T) {
 
-	gs.ResetCounters()
+		t.Skip("stream cancel returns errors normally")
 
-	m1 := make(map[string]interface{})
-	m1["name"] = "bob"
+		gs.ResetCounters()
 
-	m2 := make(map[string]interface{})
-	m2["name"] = "Kate"
+		oldData := gs.StreamData
 
-	m3 := make(map[string]interface{})
-	m3["name"] = "foo"
+		nc := 100000
+		gs.StreamData = make([]*helloworld.HelloReply, nc)
+		for i := 0; i < nc; i++ {
+			name := "name " + strconv.FormatInt(int64(i), 10)
+			gs.StreamData[i] = &helloworld.HelloReply{Message: "Hello " + name}
+		}
 
-	data := []interface{}{m1, m2, m3}
+		data := make(map[string]interface{})
+		data["name"] = "bob"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHellos",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithData(data),
+			WithInsecure(true),
+			WithName("server streaming test"),
+			WithStreamCallDuration(200*time.Millisecond),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 200*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Equal(t, "server streaming test", report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.Details)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 200*time.Millisecond && dr.Latency < 300*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		sends := gs.GetSendCounts(callType)
+		assert.NotNil(t, sends)
+		assert.Len(t, sends, 1)
+		sendCount := sends[0]
+		assert.True(t, sendCount < 20000, sendCount)
+
+		// reset
+		gs.StreamData = oldData
+	})
+
+	t.Run("with stream cancel count errors", func(t *testing.T) {
+
+		gs.ResetCounters()
+
+		oldData := gs.StreamData
+
+		nc := 100000
+		gs.StreamData = make([]*helloworld.HelloReply, nc)
+		for i := 0; i < nc; i++ {
+			name := "name " + strconv.FormatInt(int64(i), 10)
+			gs.StreamData[i] = &helloworld.HelloReply{Message: "Hello " + name}
+		}
+
+		data := make(map[string]interface{})
+		data["name"] = "bob"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHellos",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithData(data),
+			WithInsecure(true),
+			WithName("server streaming test"),
+			WithStreamCallDuration(200*time.Millisecond),
+			WithCountErrors(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 200*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Equal(t, "server streaming test", report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.Details)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Len(t, report.ErrorDist, 1)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 200*time.Millisecond && dr.Latency < 300*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		sends := gs.GetSendCounts(callType)
+		assert.NotNil(t, sends)
+		assert.Len(t, sends, 1)
+		sendCount := sends[0]
+		assert.True(t, sendCount < 30000, sendCount)
+
+		// reset
+		gs.StreamData = oldData
+	})
+
+	t.Run("with stream call count and count errors", func(t *testing.T) {
+
+		gs.ResetCounters()
+
+		oldData := gs.StreamData
+
+		nc := 5000
+		gs.StreamData = make([]*helloworld.HelloReply, nc)
+		for i := 0; i < nc; i++ {
+			name := "name " + strconv.FormatInt(int64(i), 10)
+			gs.StreamData[i] = &helloworld.HelloReply{Message: "Hello " + name}
+		}
+
+		data := make(map[string]interface{})
+		data["name"] = "bob"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHellos",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithData(data),
+			WithInsecure(true),
+			WithName("server streaming test"),
+			WithStreamCallCount(50),
+			WithCountErrors(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Equal(t, "server streaming test", report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.Details)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Len(t, report.ErrorDist, 1)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		sends := gs.GetSendCounts(callType)
+		assert.NotNil(t, sends)
+		assert.Len(t, sends, 1)
+		sendCount := sends[0]
+		assert.True(t, sendCount <= 3500, sendCount)
+
+		// reset
+		gs.StreamData = oldData
+	})
+
+	t.Run("with recv intercept", func(t *testing.T) {
+
+		gs.ResetCounters()
+
+		oldData := gs.StreamData
+
+		nc := 10000
+		gs.StreamData = make([]*helloworld.HelloReply, nc)
+		for i := 0; i < nc; i++ {
+			name := "name " + strconv.FormatInt(int64(i), 10)
+			gs.StreamData[i] = &helloworld.HelloReply{Message: "Hello " + name}
+		}
+
+		data := make(map[string]interface{})
+		data["name"] = "bob"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHellos",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithData(data),
+			WithInsecure(true),
+			WithName("server streaming test"),
+			WithCountErrors(true),
+			WithStreamRecvMsgIntercept(func(msg *dynamic.Message, err error) error {
+				if err == nil {
+					reply := &helloworld.HelloReply{}
+					convertErr := msg.ConvertTo(reply)
+					if convertErr == nil {
+						if reply.GetMessage() == "Hello name 100" {
+							return ErrEndStream
+						}
+					}
+				}
+
+				return nil
+			}),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Equal(t, "server streaming test", report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.Details)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Len(t, report.ErrorDist, 1)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		sends := gs.GetSendCounts(callType)
+		assert.NotNil(t, sends)
+		assert.Len(t, sends, 1)
+		sendCount := sends[0]
+		assert.True(t, sendCount <= 1500, sendCount)
+
+		// reset
+		gs.StreamData = oldData
+	})
+}
+
+func TestRunClientStreaming(t *testing.T) {
+	callType := helloworld.ClientStream
+
+	gs, s, err := internal.StartServer(false)
+
+	if err != nil {
+		assert.FailNow(t, err.Error())
+	}
+
+	defer s.Stop()
+
+	t.Run("basic", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+
+		data := []interface{}{m1, m2, m3}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloCS",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(16),
+			WithConcurrency(4),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.Equal(t, 16, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.NotEqual(t, report.Average, report.Slowest)
+		assert.NotEqual(t, report.Average, report.Fastest)
+		assert.NotEqual(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 16, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+	})
+
+	t.Run("with stream interval", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+
+		data := []interface{}{m1, m2, m3, m4}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloCS",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(200*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 600*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 600*time.Millisecond && dr.Latency < 800*time.Millisecond, dr.Latency.String()+" not in interval")
+	})
+
+	t.Run("with stream cancel", func(t *testing.T) {
+		gs.ResetCounters()
+
+		nc := 100000
+		data := make([]interface{}, nc)
+		for i := 0; i < nc; i++ {
+			data[i] = map[string]interface{}{
+				"name": "name " + strconv.FormatInt(int64(i), 10),
+			}
+		}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloCS",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamCallDuration(200*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 200*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 200*time.Millisecond && dr.Latency < 400*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.True(t, len(msgs) < 15000, len(msgs))
+	})
+
+	t.Run("with stream interval and cancel", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+		m5 := make(map[string]interface{})
+		m5["name"] = "biz"
+
+		data := []interface{}{m1, m2, m3, m4, m5}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloCS",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(200*time.Millisecond),
+			WithStreamCallDuration(650*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 650*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 650*time.Millisecond && dr.Latency < 800*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 4)
+	})
+
+	t.Run("with stream interval and call count", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+		m5 := make(map[string]interface{})
+		m5["name"] = "biz"
+
+		data := []interface{}{m1, m2, m3, m4, m5}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloCS",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithStreamCallCount(3),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 200*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		// allow extra time for last message send
+		assert.True(t, dr.Latency > 200*time.Millisecond && dr.Latency < 400*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 3)
+	})
+
+	t.Run("with stream dynamic message true", func(t *testing.T) {
+		gs.ResetCounters()
+
+		data := make(map[string]interface{})
+		data["name"] = "{{newUUID}}"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloCS",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithStreamCallCount(4),
+			WithStreamDynamicMessages(true),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 300*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		// allow 100ms for each dynamic message (re)generation
+		assert.True(t, dr.Latency > 300*time.Millisecond && dr.Latency < 700*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 4)
+		for i, m1 := range msgs {
+			parsed, err := uuid.Parse(m1.GetName())
+			assert.NoError(t, err)
+			assert.NotEmpty(t, parsed)
+
+			for j, m2 := range msgs {
+				parsed, err = uuid.Parse(m2.GetName())
+				assert.NoError(t, err)
+				assert.NotEmpty(t, parsed)
+
+				if i != j {
+					assert.NotEqual(t, m1.GetName(), m2.GetName())
+				}
+			}
+		}
+	})
+
+	t.Run("with stream dynamic message false", func(t *testing.T) {
+		gs.ResetCounters()
+
+		data := make(map[string]interface{})
+		data["name"] = "{{newUUID}}"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloCS",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithStreamCallCount(4),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 300*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 300*time.Millisecond && dr.Latency < 400*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 4)
+		for i, m1 := range msgs {
+			parsed, err := uuid.Parse(m1.GetName())
+			assert.NoError(t, err)
+			assert.NotEmpty(t, parsed)
+
+			for j, m2 := range msgs {
+				parsed, err = uuid.Parse(m2.GetName())
+				assert.NoError(t, err)
+				assert.NotEmpty(t, parsed)
+
+				if i != j {
+					assert.Equal(t, m1.GetName(), m2.GetName())
+				}
+			}
+		}
+	})
+}
+
+func TestRunClientStreamingBinary(t *testing.T) {
+	callType := helloworld.ClientStream
+
+	gs, s, err := internal.StartServer(false)
+
+	if err != nil {
+		assert.FailNow(t, err.Error())
+	}
+
+	defer s.Stop()
+
+	gs.ResetCounters()
+
+	msg := &helloworld.HelloRequest{}
+	msg.Name = "bob"
+
+	binData, err := proto.Marshal(msg)
+	assert.NoError(t, err)
 
 	report, err := Run(
-		"helloworld.Greeter.SayHelloBidi",
+		"helloworld.Greeter.SayHelloCS",
 		internal.TestLocalhost,
 		WithProtoFile("../testdata/greeter.proto", []string{}),
-		WithTotalRequests(20),
+		WithTotalRequests(24),
 		WithConcurrency(4),
 		WithTimeout(time.Duration(20*time.Second)),
 		WithDialTimeout(time.Duration(20*time.Second)),
-		WithData(data),
+		WithBinaryData(binData),
 		WithInsecure(true),
 	)
 
-	assert.NoError(t, err)
+	assert.NoError(t, err)
+
+	assert.NotNil(t, report)
+
+	assert.Equal(t, 24, int(report.Count))
+	assert.NotZero(t, report.Average)
+	assert.NotZero(t, report.Fastest)
+	assert.NotZero(t, report.Slowest)
+	assert.NotZero(t, report.Rps)
+	assert.Empty(t, report.Name)
+	assert.NotEmpty(t, report.Date)
+	assert.NotEmpty(t, report.Details)
+	assert.NotEmpty(t, report.Options)
+	assert.Equal(t, true, report.Options.Insecure)
+	assert.NotEmpty(t, report.LatencyDistribution)
+	assert.Equal(t, ReasonNormalEnd, report.EndReason)
+	assert.Empty(t, report.ErrorDist)
+
+	assert.NotEqual(t, report.Average, report.Slowest)
+	assert.NotEqual(t, report.Average, report.Fastest)
+	assert.NotEqual(t, report.Slowest, report.Fastest)
+
+	count := gs.GetCount(callType)
+	assert.Equal(t, 24, count)
+
+	connCount := gs.GetConnectionCount()
+	assert.Equal(t, 1, connCount)
+}
+
+func TestRunBidi(t *testing.T) {
+	callType := helloworld.Bidi
+
+	gs, s, err := internal.StartServer(false)
+
+	if err != nil {
+		assert.FailNow(t, err.Error())
+	}
+
+	defer s.Stop()
+
+	t.Run("basic", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+
+		data := []interface{}{m1, m2, m3}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(20),
+			WithConcurrency(4),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.Equal(t, 20, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.NotEqual(t, report.Average, report.Slowest)
+		assert.NotEqual(t, report.Average, report.Fastest)
+		assert.NotEqual(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 20, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+	})
+
+	t.Run("with stream interval", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+
+		data := []interface{}{m1, m2, m3, m4}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(200*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 600*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		// allow extra for the last send and receive
+		assert.True(t, dr.Latency > 600*time.Millisecond && dr.Latency < 950*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+	})
+
+	t.Run("with stream cancel", func(t *testing.T) {
+		gs.ResetCounters()
+
+		nc := 100000
+		data := make([]interface{}, nc)
+		for i := 0; i < nc; i++ {
+			data[i] = map[string]interface{}{
+				"name": "name " + strconv.FormatInt(int64(i), 10),
+			}
+		}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamCallDuration(200*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 200*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 200*time.Millisecond && dr.Latency < 1000*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.True(t, len(msgs) < 15000, len(msgs))
+	})
+
+	t.Run("with stream interval and cancel", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+		m2 := make(map[string]interface{})
+		m2["name"] = "kate"
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+		m5 := make(map[string]interface{})
+		m5["name"] = "biz"
+		m6 := make(map[string]interface{})
+		m6["name"] = "baz"
+		m7 := make(map[string]interface{})
+		m7["name"] = "alice"
+
+		data := []interface{}{m1, m2, m3, m4, m5, m6, m7}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(200*time.Millisecond),
+			WithStreamCallDuration(620*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 620*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 620*time.Millisecond && dr.Latency < 820*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 4)
+	})
+
+	t.Run("with stream count", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+		m5 := make(map[string]interface{})
+		m5["name"] = "biz"
+
+		data := []interface{}{m1, m2, m3, m4, m5}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithStreamCallCount(3),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 200*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.Empty(t, report.ErrorDist)
 
-	assert.NotNil(t, report)
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
 
-	assert.Equal(t, 20, int(report.Count))
-	assert.NotZero(t, report.Average)
-	assert.NotZero(t, report.Fastest)
-	assert.NotZero(t, report.Slowest)
-	assert.NotZero(t, report.Rps)
-	assert.Empty(t, report.Name)
-	assert.NotEmpty(t, report.Date)
-	assert.NotEmpty(t, report.Details)
-	assert.NotEmpty(t, report.Options)
-	assert.NotEmpty(t, report.LatencyDistribution)
-	assert.Equal(t, ReasonNormalEnd, report.EndReason)
-	assert.Equal(t, true, report.Options.Insecure)
-	assert.Empty(t, report.ErrorDist)
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
 
-	assert.NotEqual(t, report.Average, report.Slowest)
-	assert.NotEqual(t, report.Average, report.Fastest)
-	assert.NotEqual(t, report.Slowest, report.Fastest)
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
 
-	count := gs.GetCount(callType)
-	assert.Equal(t, 20, count)
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 200*time.Millisecond && dr.Latency < 300*time.Millisecond, dr.Latency.String()+" not in interval")
 
-	connCount := gs.GetConnectionCount()
-	assert.Equal(t, 1, connCount)
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 3)
+	})
+
+	t.Run("with stream dynamic message true", func(t *testing.T) {
+		gs.ResetCounters()
+
+		data := make(map[string]interface{})
+		data["name"] = "{{newUUID}}"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithStreamCallCount(4),
+			WithStreamDynamicMessages(true),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 300*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 300*time.Millisecond && dr.Latency < 400*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 4)
+		for i, m1 := range msgs {
+			parsed, err := uuid.Parse(m1.GetName())
+			assert.NoError(t, err)
+			assert.NotEmpty(t, parsed)
+
+			for j, m2 := range msgs {
+				parsed, err = uuid.Parse(m2.GetName())
+				assert.NoError(t, err)
+				assert.NotEmpty(t, parsed)
+
+				if i != j {
+					assert.NotEqual(t, m1.GetName(), m2.GetName())
+				}
+			}
+		}
+	})
+
+	t.Run("with stream dynamic message false", func(t *testing.T) {
+		gs.ResetCounters()
+
+		data := make(map[string]interface{})
+		data["name"] = "{{newUUID}}"
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithStreamCallCount(4),
+			WithData(data),
+			WithInsecure(true),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 300*time.Millisecond, report.Total.String()+" not in interval")
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 300*time.Millisecond && dr.Latency < 400*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 4)
+		for i, m1 := range msgs {
+			parsed, err := uuid.Parse(m1.GetName())
+			assert.NoError(t, err)
+			assert.NotEmpty(t, parsed)
+
+			for j, m2 := range msgs {
+				parsed, err = uuid.Parse(m2.GetName())
+				assert.NoError(t, err)
+				assert.NotEmpty(t, parsed)
+
+				if i != j {
+					assert.Equal(t, m1.GetName(), m2.GetName())
+				}
+			}
+		}
+	})
+
+	t.Run("with stream recv message", func(t *testing.T) {
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+		m5 := make(map[string]interface{})
+		m5["name"] = "biz"
+		m6 := make(map[string]interface{})
+		m6["name"] = "baz"
+
+		data := []interface{}{m1, m2, m3, m4, m5, m6}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+			WithStreamRecvMsgIntercept(func(msg *dynamic.Message, err error) error {
+				if err == nil {
+					reply := &helloworld.HelloReply{}
+					convertErr := msg.ConvertTo(reply)
+					if convertErr == nil {
+						if reply.GetMessage() == "Hello bar" {
+							return ErrEndStream
+						}
+					}
+				}
+
+				return nil
+			}),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 300*time.Millisecond)
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 300*time.Millisecond && dr.Latency < 400*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 4)
+	})
+
+	t.Run("with stream recv message no stop", func(t *testing.T) {
+
+		gs.ResetCounters()
+
+		m1 := make(map[string]interface{})
+		m1["name"] = "bob"
+		m2 := make(map[string]interface{})
+		m2["name"] = "Kate"
+		m3 := make(map[string]interface{})
+		m3["name"] = "foo"
+		m4 := make(map[string]interface{})
+		m4["name"] = "bar"
+		m5 := make(map[string]interface{})
+		m5["name"] = "biz"
+		m6 := make(map[string]interface{})
+		m6["name"] = "baz"
+
+		data := []interface{}{m1, m2, m3, m4, m5, m6}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHelloBidi",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(1),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithStreamInterval(100*time.Millisecond),
+			WithData(data),
+			WithInsecure(true),
+			WithStreamRecvMsgIntercept(func(msg *dynamic.Message, err error) error {
+				if err == nil {
+					reply := &helloworld.HelloReply{}
+					convertErr := msg.ConvertTo(reply)
+					if convertErr == nil {
+						// bat doesn't exist
+						if reply.GetMessage() == "Hello bat" {
+							return ErrEndStream
+						}
+					}
+				}
+
+				return nil
+			}),
+		)
+
+		assert.NoError(t, err)
+
+		assert.NotNil(t, report)
+
+		assert.True(t, report.Total > 500*time.Millisecond)
+		assert.Equal(t, 1, int(report.Count))
+		assert.NotZero(t, report.Average)
+		assert.NotZero(t, report.Fastest)
+		assert.NotZero(t, report.Slowest)
+		assert.NotZero(t, report.Rps)
+		assert.Empty(t, report.Name)
+		assert.NotEmpty(t, report.Date)
+		assert.NotEmpty(t, report.Details)
+		assert.NotEmpty(t, report.Options)
+		assert.NotEmpty(t, report.LatencyDistribution)
+		assert.Equal(t, ReasonNormalEnd, report.EndReason)
+		assert.Equal(t, true, report.Options.Insecure)
+		assert.Empty(t, report.ErrorDist)
+
+		assert.Equal(t, report.Average, report.Slowest)
+		assert.Equal(t, report.Average, report.Fastest)
+		assert.Equal(t, report.Slowest, report.Fastest)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 1, count)
+
+		connCount := gs.GetConnectionCount()
+		assert.Equal(t, 1, connCount)
+
+		assert.Len(t, report.Details, 1)
+		dr := report.Details[0]
+		assert.True(t, dr.Latency > 500*time.Millisecond && dr.Latency < 600*time.Millisecond, dr.Latency.String()+" not in interval")
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 1)
+		msgs := calls[0]
+		assert.Len(t, msgs, 6)
+	})
 }
 
 func TestRunUnarySecure(t *testing.T) {
+
+	t.Skip("asdf")
+
 	callType := helloworld.Unary
 
 	gs, s, err := internal.StartServer(true)