Repository: honeydipper/honeydipper
Pull request #528 (2023-03-23 21:19:56)

refactor: abstract comm I/O with MessageReceiver 
#### Description
Make comm API and driver API abstract for future extensibility and test-ability.

#### This PR fixes the following issues
N/A


Patch:
diff --git a/internal/api/response.go b/internal/api/response.go
index 38f6f0e..a513f09 100644
--- a/internal/api/response.go
+++ b/internal/api/response.go
@@ -8,7 +8,6 @@ package api
 
 import (
 	"fmt"
-	"io"
 	"strconv"
 
 	"github.com/honeydipper/honeydipper/pkg/dipper"
@@ -22,14 +21,14 @@ const DefaultAPILockExpireMS = 1000
 
 // Response is used for responding to the api service.
 type Response struct {
-	EventBus io.Writer
+	EventBus dipper.MessageReceiver
 	Request  *dipper.Message
 	Acked    bool
 }
 
 // Ack acks a call.
 func (resp *Response) Ack() {
-	dipper.SendMessage(resp.EventBus, &dipper.Message{
+	resp.EventBus.SendMessage(&dipper.Message{
 		Channel: "eventbus",
 		Subject: "api",
 		Labels: map[string]string{
@@ -43,7 +42,7 @@ func (resp *Response) Ack() {
 
 // Return returns data to api service.
 func (resp *Response) Return(data interface{}) {
-	dipper.SendMessage(resp.EventBus, &dipper.Message{
+	resp.EventBus.SendMessage(&dipper.Message{
 		Channel: "eventbus",
 		Subject: "api",
 		Labels: map[string]string{
@@ -57,7 +56,7 @@ func (resp *Response) Return(data interface{}) {
 
 // ReturnError returns an error to the API service.
 func (resp *Response) ReturnError(err error) {
-	dipper.SendMessage(resp.EventBus, &dipper.Message{
+	resp.EventBus.SendMessage(&dipper.Message{
 		Channel: "eventbus",
 		Subject: "api",
 		Labels: map[string]string{
@@ -94,7 +93,7 @@ func NewResponseFactory() *ResponseFactory {
 }
 
 // NewResponse provides a function to create new api Response.
-func (rf *ResponseFactory) NewResponse(caller dipper.RPCCaller, eventbus io.Writer, m *dipper.Message) *Response {
+func (rf *ResponseFactory) NewResponse(caller dipper.RPCCaller, eventbus dipper.MessageReceiver, m *dipper.Message) *Response {
 	resp := &Response{
 		EventBus: eventbus,
 		Request:  m,
diff --git a/internal/driver/builtin.go b/internal/driver/builtin.go
index 1ea7bd4..6b68e14 100644
--- a/internal/driver/builtin.go
+++ b/internal/driver/builtin.go
@@ -8,20 +8,33 @@ package driver
 
 import (
 	"fmt"
+	"io"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strings"
+
+	"github.com/honeydipper/honeydipper/internal/daemon"
+	"github.com/honeydipper/honeydipper/pkg/dipper"
 )
 
 // BuiltinDriver are compiled and delivered with daemon binary in the same container image.
 type BuiltinDriver struct {
-	meta *Meta
+	meta   *Meta
+	stream chan<- *dipper.Message
+
+	input  io.ReadCloser
+	output io.WriteCloser
+	run    *exec.Cmd
 }
 
 // BuiltinPath is the path where the builtin drivers are kept. It will try using $HONEYDIPPER_DRIVERS_BUILTIN by default.
 // If $HONEYDIPPER_DRIVERS_BUILTIN is not set, will try using $GOPATH/bin.  If $GOPATH is not defined, use "/opt/honeydipper/driver/builtin".
 var BuiltinPath string
 
+// replace the func variable with mock during testing.
+var execCommand = exec.Command
+
 func builtinPath() string {
 	if BuiltinPath == "" {
 		driverPath, ok := os.LookupEnv("HONEYDIPPER_DRIVERS_BUILTIN")
@@ -53,7 +66,9 @@ func (d *BuiltinDriver) Acquire() {
 
 // Prepare function is used for preparing the arguments when calling the executable.
 // for the driver.
-func (d *BuiltinDriver) Prepare() {
+func (d *BuiltinDriver) Prepare(stream chan<- *dipper.Message) {
+	d.stream = stream
+
 	var argsList []interface{}
 
 	args, ok := d.meta.HandlerData["arguments"]
@@ -82,3 +97,70 @@ func (d *BuiltinDriver) Meta() *Meta {
 func NewBuiltinDriver(m *Meta) *BuiltinDriver {
 	return &BuiltinDriver{meta: m}
 }
+
+// Start the driver child process.  The "service" indicates which service this driver belongs to.
+func (d *BuiltinDriver) Start(service string) {
+	d.run = execCommand(d.meta.Executable, append([]string{service}, d.meta.Arguments...)...)
+	if input, err := d.run.StdoutPipe(); err != nil {
+		dipper.Logger.Panicf("[%s] Unable to link to driver stdout %v", service, err)
+	} else {
+		d.input = input
+		go d.fetchMessages(service)
+	}
+
+	if output, err := d.run.StdinPipe(); err != nil {
+		dipper.Logger.Panicf("[%s] Unable to link to driver stdin %v", service, err)
+	} else {
+		d.output = output
+	}
+
+	d.run.Stderr = os.Stderr
+	d.run.ExtraFiles = []*os.File{os.Stdout} // giving child process stdout for logging
+	if err := d.run.Start(); err != nil {
+		dipper.Logger.Panicf("[%s] Failed to start driver %v", service, err)
+	}
+}
+
+// SendMessage sends a dipper message to the driver child process.
+func (d *BuiltinDriver) SendMessage(msg *dipper.Message) {
+	dipper.SendMessage(d.output, msg)
+}
+
+func (d *BuiltinDriver) fetchMessages(service string) {
+	quit := false
+	daemon.Children.Add(1)
+	defer daemon.Children.Done()
+	for !quit && !daemon.ShuttingDown {
+		func() {
+			defer dipper.SafeExitOnError(
+				"failed to fetching messages from driver %s.%s",
+				service,
+				d.meta.Name,
+			)
+			defer dipper.CatchError(io.EOF, func() { quit = true })
+			for !quit && !daemon.ShuttingDown {
+				message := dipper.FetchRawMessage(d.input)
+				d.stream <- message
+			}
+		}()
+	}
+	dipper.Logger.Warningf("[%s-%s] driver closed for business", service, d.meta.Name)
+}
+
+// Close close all the channels for the driver and signal the child process to exit.
+func (d *BuiltinDriver) Close() {
+	if d.stream != nil {
+		close(d.stream)
+		d.stream = nil
+	}
+	if d.output != nil {
+		d.output.Close()
+		d.output = nil
+	}
+}
+
+// Wait wait for the driver process to exit.
+func (d *BuiltinDriver) Wait() {
+	_ = d.run.Wait()
+	d.run = nil
+}
diff --git a/internal/driver/driver.go b/internal/driver/driver.go
index 69e3174..25486de 100644
--- a/internal/driver/driver.go
+++ b/internal/driver/driver.go
@@ -9,9 +9,7 @@ package driver
 import (
 	"errors"
 	"fmt"
-	"io"
-	"os"
-	"os/exec"
+	"time"
 
 	"github.com/honeydipper/honeydipper/internal/daemon"
 	"github.com/honeydipper/honeydipper/pkg/dipper"
@@ -27,19 +25,31 @@ const DriverMessageBuffer = 10
 // Handler provides common functions for handling a driver.
 type Handler interface {
 	Acquire()
-	Prepare()
 	Meta() *Meta
+	Prepare(input chan<- *dipper.Message)
+	SendMessage(*dipper.Message)
+	Start(string)
+	Close()
+	Wait()
 }
 
-// replace the func variable with mock during testing.
-var execCommand = exec.Command
+// Runtime contains the runtime information of the running driver.
+type Runtime struct {
+	Data        interface{}
+	DynamicData interface{}
+	Feature     string
+	Handler     Handler
+	Stream      <-chan *dipper.Message
+	Service     string
+	State       int
+}
 
 // NewDriver creates a driver object to represent a child process.
-func NewDriver(data map[string]interface{}) Handler {
+func NewDriver(feature string, metaData map[string]interface{}, driverData interface{}, dynamicData interface{}) *Runtime {
 	var meta Meta
-	err := mapstructure.Decode(data, &meta)
+	err := mapstructure.Decode(metaData, &meta)
 	if err != nil {
-		panic(fmt.Errorf("malformat driver meta: %+v: %w", data, err))
+		panic(fmt.Errorf("malformat driver meta: %+v: %w", metaData, err))
 	}
 
 	if meta.Name == "" {
@@ -51,62 +61,32 @@ func NewDriver(data map[string]interface{}) Handler {
 	switch meta.Type {
 	case "builtin":
 		dh = NewBuiltinDriver(&meta)
+	case "null":
+		dh = NewNullDriver(&meta)
 	default:
 		panic(fmt.Errorf("%w: unsupported driver type: %s", ErrDriverError, meta.Type))
 	}
 
-	dh.Acquire()
-	dh.Prepare()
+	stream := make(chan *dipper.Message, DriverMessageBuffer)
 
-	if meta.Executable == "" {
-		panic(fmt.Errorf("%w: executable not defined for driver: %s", ErrDriverError, meta.Name))
+	runtime := &Runtime{
+		Data:        driverData,
+		DynamicData: dynamicData,
+		Feature:     feature,
+		Handler:     dh,
+		Stream:      stream,
 	}
 
-	return dh
-}
+	dh.Acquire()
+	dh.Prepare(stream)
 
-// Runtime contains the runtime information of the running driver.
-type Runtime struct {
-	Data        interface{}
-	DynamicData interface{}
-	Feature     string
-	Stream      chan dipper.Message
-	Input       io.ReadCloser
-	Output      io.WriteCloser
-	Service     string
-	Run         *exec.Cmd
-	State       int
-	Handler     Handler
+	return runtime
 }
 
 // Start the driver child process.  The "service" indicates which service this driver belongs to.
 func (runtime *Runtime) Start(service string) {
 	runtime.Service = service
-
-	m := runtime.Handler.Meta()
-
-	run := execCommand(m.Executable, append([]string{service}, m.Arguments...)...)
-	if input, err := run.StdoutPipe(); err != nil {
-		dipper.Logger.Panicf("[%s] Unable to link to driver stdout %v", service, err)
-	} else {
-		runtime.Input = input
-		runtime.Stream = make(chan dipper.Message, DriverMessageBuffer)
-		go runtime.fetchMessages()
-	}
-
-	if output, err := run.StdinPipe(); err != nil {
-		dipper.Logger.Panicf("[%s] Unable to link to driver stdin %v", service, err)
-	} else {
-		runtime.Output = output
-	}
-
-	run.Stderr = os.Stderr
-	run.ExtraFiles = []*os.File{os.Stdout} // giving child process stdout for logging
-	if err := run.Start(); err != nil {
-		dipper.Logger.Panicf("[%s] Failed to start driver %v", service, err)
-	}
-
-	runtime.Run = run
+	runtime.Handler.Start(service)
 	runtime.SendOptions()
 }
 
@@ -127,7 +107,7 @@ func (runtime *Runtime) SendOptions() {
 	})
 }
 
-// SendMessage sentds a dipper message to the driver child process.
+// SendMessage sends a dipper message to the driver child process.
 func (runtime *Runtime) SendMessage(msg *dipper.Message) {
 	if runtime.Feature != "emitter" {
 		if emitter, ok := daemon.Emitters[runtime.Service]; ok {
@@ -140,26 +120,18 @@ func (runtime *Runtime) SendMessage(msg *dipper.Message) {
 			})
 		}
 	}
-	dipper.SendMessage(runtime.Output, msg)
+	runtime.Handler.SendMessage(msg)
 }
 
-func (runtime *Runtime) fetchMessages() {
-	quit := false
-	daemon.Children.Add(1)
-	defer daemon.Children.Done()
-	for !quit && !daemon.ShuttingDown {
-		func() {
-			defer dipper.SafeExitOnError(
-				"failed to fetching messages from driver %s.%s",
-				runtime.Service,
-				runtime.Handler.Meta().Name,
-			)
-			defer dipper.CatchError(io.EOF, func() { quit = true })
-			for !quit && !daemon.ShuttingDown {
-				message := dipper.FetchRawMessage(runtime.Input)
-				runtime.Stream <- *message
-			}
-		}()
+// Ready waits until the driver is alive or report error.
+func (runtime *Runtime) Ready(d time.Duration) {
+	var elapsed time.Duration
+	for runtime.State < DriverAlive && elapsed < d {
+		time.Sleep(time.Second)
+		elapsed += time.Second
+	}
+
+	if runtime.State != DriverAlive {
+		panic(fmt.Errorf("%w: feature failed or loading timeout: %s", ErrDriverError, runtime.Feature))
 	}
-	dipper.Logger.Warningf("[%s-%s] driver closed for business", runtime.Service, runtime.Handler.Meta().Name)
 }
diff --git a/internal/driver/nulldriver.go b/internal/driver/nulldriver.go
new file mode 100644
index 0000000..ab49c70
--- /dev/null
+++ b/internal/driver/nulldriver.go
@@ -0,0 +1,82 @@
+// Copyright 2023 PayPal Inc.
+
+// This Source Code Form is subject to the terms of the MIT License.
+// If a copy of the MIT License was not distributed with this file,
+// you can obtain one at https://mit-license.org/.
+
+package driver
+
+import (
+	"github.com/honeydipper/honeydipper/pkg/dipper"
+)
+
+// NullDriverHandler provides support for null driver.
+type NullDriverHandler struct {
+	AcquireFunc     func()
+	MetaFunc        func() *Meta
+	PrepareFunc     func(input chan<- *dipper.Message)
+	SendMessageFunc func(*dipper.Message)
+	StartFunc       func(string)
+	CloseFunc       func()
+	WaitFunc        func()
+}
+
+// Acquire does nothing unless overridden with AcquireFunc.
+func (h *NullDriverHandler) Acquire() {
+	if h.AcquireFunc != nil {
+		h.AcquireFunc()
+	}
+}
+
+// Meta does nothing unless overridden with MetaFunc.
+func (h *NullDriverHandler) Meta() *Meta {
+	if h.MetaFunc != nil {
+		return h.MetaFunc()
+	}
+
+	return nil
+}
+
+// Prepare does nothing unless overridden with PrepareFunc.
+func (h *NullDriverHandler) Prepare(stream chan<- *dipper.Message) {
+	if h.PrepareFunc != nil {
+		h.PrepareFunc(stream)
+	}
+}
+
+// Start does nothing unless overridden with StartFunc.
+func (h *NullDriverHandler) Start(service string) {
+	if h.StartFunc != nil {
+		h.StartFunc(service)
+	}
+}
+
+// SendMessage does nothing unless overridden with SendMessageFunc.
+func (h *NullDriverHandler) SendMessage(msg *dipper.Message) {
+	if h.SendMessageFunc != nil {
+		h.SendMessageFunc(msg)
+	}
+}
+
+// Close does nothing unless overridden with CloseFunc.
+func (h *NullDriverHandler) Close() {
+	if h.CloseFunc != nil {
+		h.CloseFunc()
+	}
+}
+
+// Wait does nothing unless overridden with WaitFunc.
+func (h *NullDriverHandler) Wait() {
+	if h.WaitFunc != nil {
+		h.WaitFunc()
+	}
+}
+
+// NewNullDriver creates a null driver handler.
+func NewNullDriver(meta *Meta) *NullDriverHandler {
+	return &NullDriverHandler{
+		MetaFunc: func() *Meta {
+			return meta
+		},
+	}
+}
diff --git a/internal/service/engine.go b/internal/service/engine.go
index 94a7406..fab4cc3 100644
--- a/internal/service/engine.go
+++ b/internal/service/engine.go
@@ -40,7 +40,7 @@ type WorkflowHelper struct {
 // SendMessage method sends workflow messages to eventbus channle.
 func (h *WorkflowHelper) SendMessage(msg *dipper.Message) {
 	worker := h.engine.getDriverRuntime(dipper.ChannelEventbus)
-	dipper.SendMessage(worker.Output, msg)
+	worker.SendMessage(msg)
 }
 
 // GetConfig method feed config from service to workflow engine.
diff --git a/internal/service/operator.go b/internal/service/operator.go
index 5a26a40..2bdb091 100644
--- a/internal/service/operator.go
+++ b/internal/service/operator.go
@@ -40,7 +40,7 @@ func handleEventbusCommand(msg *dipper.Message) []RoutedMessage {
 				newLabels["status"] = "error"
 				newLabels["reason"] = fmt.Sprintf("%+v", r)
 				eventbus := operator.getDriverRuntime(dipper.ChannelEventbus)
-				dipper.SendMessage(eventbus.Output, &dipper.Message{
+				eventbus.SendMessage(&dipper.Message{
 					Channel: dipper.ChannelEventbus,
 					Subject: dipper.EventbusReturn,
 					Labels:  newLabels,
diff --git a/internal/service/service.go b/internal/service/service.go
index fe39abb..21c80f0 100644
--- a/internal/service/service.go
+++ b/internal/service/service.go
@@ -9,7 +9,6 @@ package service
 import (
 	"errors"
 	"fmt"
-	"io"
 	"os"
 	"reflect"
 	"strconv"
@@ -129,23 +128,15 @@ func (s *Service) GetName() string {
 	return s.name
 }
 
-// GetStream returns the writer for communication with the specified feature.
-func (s *Service) GetStream(feature string) io.Writer {
-	if runtime := s.getDriverRuntime(feature); runtime != nil {
-		seconds := 0
-		for seconds < 9 && (runtime.State == driver.DriverLoading || runtime.State == driver.DriverReloading) {
-			dipper.Logger.Warningf("[%s] waiting for feature %s to load/reload ...", s.name, feature)
-			time.Sleep(time.Second)
-			seconds++
-		}
-
-		if runtime.State != driver.DriverAlive {
-			panic(fmt.Errorf("%w: feature failed or loading timeout: %s", ErrServiceError, feature))
-		}
-
-		return runtime.Output
+// GetReceiver returns the driver object that receives the rpc messages.
+func (s *Service) GetReceiver(feature string) interface{} {
+	receiver := s.getDriverRuntime(feature)
+	if receiver == nil {
+		panic(fmt.Errorf("%w: feature not loaded: %s", ErrServiceError, feature))
 	}
-	panic(fmt.Errorf("%w: feature not loaded: %s", ErrServiceError, feature))
+	receiver.Ready(DriverReadyTimeout * time.Second)
+
+	return receiver
 }
 
 func (s *Service) loadFeature(feature string) (affected bool, driverName string, rerr error) {
@@ -192,34 +183,29 @@ func (s *Service) loadFeature(feature string) (affected bool, driverName string,
 		dynamicData = s.dynamicFeatureData[feature]
 	}
 
-	var driverHandler driver.Handler
-	if driverMeta, ok := s.config.GetStagedDriverData(fmt.Sprintf("daemon.drivers.%s", driverName)); ok {
-		driverHandler = driver.NewDriver(driverMeta.(map[string]interface{}))
-	} else {
+	driverMeta, ok := s.config.GetStagedDriverData(fmt.Sprintf("daemon.drivers.%s", driverName))
+	if !ok {
 		panic("unable to get driver metadata")
 	}
 
-	dipper.Logger.Debugf("[%s] driver %s meta %v", s.name, driverName, driverHandler.Meta())
-	driverRuntime := driver.Runtime{
-		Feature:     feature,
-		Data:        driverData,
-		DynamicData: dynamicData,
-		State:       driver.DriverLoading,
-		Handler:     driverHandler,
-	}
+	driverRuntime := driver.NewDriver(feature, driverMeta.(map[string]interface{}), driverData, dynamicData)
+	dipper.Logger.Debugf("[%s] driver %s meta %v", s.name, driverName, driverRuntime.Handler.Meta())
+
+	driverMetaUnchanged := oldRuntime != nil && reflect.DeepEqual(*oldRuntime.Handler.Meta(), *driverRuntime.Handler.Meta())
+	driverRunning := oldRuntime != nil && oldRuntime.State != driver.DriverFailed
 
-	if oldRuntime != nil && oldRuntime.State != driver.DriverFailed && reflect.DeepEqual(*oldRuntime.Handler.Meta(), *driverHandler.Meta()) {
+	if driverRunning && driverMetaUnchanged {
 		if reflect.DeepEqual(oldRuntime.Data, driverRuntime.Data) && reflect.DeepEqual(oldRuntime.DynamicData, driverRuntime.DynamicData) {
 			dipper.Logger.Infof("[%s] driver not affected: %s", s.name, driverName)
 		} else {
 			// hot reload
 			affected = true
-			s.hotReload(&driverRuntime, oldRuntime)
+			s.hotReload(driverRuntime, oldRuntime)
 		}
 	} else {
 		// cold reload
 		affected = true
-		s.coldReload(&driverRuntime, oldRuntime)
+		s.coldReload(driverRuntime, oldRuntime)
 	}
 
 	return affected, driverName, nil
@@ -239,10 +225,9 @@ func (s *Service) coldReload(driverRuntime *driver.Runtime, oldRuntime *driver.R
 	go func(s *Service, runtime *driver.Runtime) {
 		defer dipper.SafeExitOnError("[%s] driver runtime %s crash", s.name, runtime.Handler.Meta().Name)
 		defer s.checkDeleteDriverRuntime(runtime.Feature, runtime)
-		defer close(runtime.Stream)
+		defer runtime.Handler.Close()
 
-		//nolint:errcheck
-		runtime.Run.Wait()
+		runtime.Handler.Wait()
 	}(s, driverRuntime)
 
 	if oldRuntime != nil {
@@ -255,7 +240,7 @@ func (s *Service) coldReload(driverRuntime *driver.Runtime, oldRuntime *driver.R
 			defer dipper.SafeExitOnError("[%s] runtime %s being replaced output is already closed", s.name, runtime.Handler.Meta().Name)
 			// allow 50 millisecond for the data to drain
 			time.Sleep(DriverGracefulTimeout * time.Millisecond)
-			runtime.Output.Close()
+			runtime.Handler.Close()
 		}(oldRuntime)
 	}
 }
@@ -338,7 +323,7 @@ func (s *Service) removeUnusedFeatures(featureList map[string]bool) {
 				defer dipper.SafeExitOnError("[%s] unused runtime %s output is already closed", s.name, runtime.Handler.Meta().Name)
 				// allow 50 millisecond for the data to drain
 				time.Sleep(DriverGracefulTimeout * time.Millisecond)
-				runtime.Output.Close()
+				runtime.Handler.Close()
 			}(runtime)
 		}
 	}
@@ -468,7 +453,7 @@ func (s *Service) serviceLoop() {
 			func() {
 				defer dipper.SafeExitOnError("[%s] service loop continue", s.name)
 				runtime := orderedRuntimes[chosen]
-				msg := value.Interface().(dipper.Message)
+				msg := value.Interface().(*dipper.Message)
 				if runtime.Feature != FeatureEmitter {
 					if emitter, ok := daemon.Emitters[s.name]; ok {
 						emitter.CounterIncr("honey.honeydipper.local.message", []string{
@@ -483,7 +468,7 @@ func (s *Service) serviceLoop() {
 
 				s.driverLock.Lock()
 				defer s.driverLock.Unlock()
-				go s.process(msg, runtime)
+				go s.process(*msg, runtime)
 			}()
 
 		case !ok && chosen < len(orderedRuntimes):
@@ -505,7 +490,7 @@ func (s *Service) serviceLoop() {
 	for fname, runtime := range s.driverRuntimes {
 		func() {
 			defer dipper.SafeExitOnError("[%s] driver runtime for feature %s already closed", s.name, fname)
-			runtime.Output.Close()
+			runtime.Handler.Close()
 		}()
 	}
 	dipper.Logger.Warningf("[%s] service closed for business", s.name)
@@ -636,7 +621,7 @@ func (s *Service) checkDeleteDriverRuntime(feature string, check *driver.Runtime
 func coldReloadDriverRuntime(d *driver.Runtime, m *dipper.Message) {
 	s := Services[d.Service]
 	s.checkDeleteDriverRuntime(d.Feature, d)
-	d.Output.Close()
+	d.Handler.Close()
 	dipper.Must(s.loadFeature(d.Feature))
 }
 
@@ -684,7 +669,7 @@ func handleRPCCall(from *driver.Runtime, m *dipper.Message) {
 	feature := m.Labels["feature"]
 	m.Labels["caller"] = from.Feature
 	s := Services[from.Service]
-	dipper.SendMessage(s.getDriverRuntime(feature).Output, m)
+	s.getDriverRuntime(feature).SendMessage(m)
 }
 
 func handleRPCReturn(from *driver.Runtime, m *dipper.Message) {
@@ -693,14 +678,14 @@ func handleRPCReturn(from *driver.Runtime, m *dipper.Message) {
 	if caller == "-" {
 		s.HandleReturn(m)
 	} else {
-		dipper.SendMessage(s.getDriverRuntime(caller).Output, m)
+		s.getDriverRuntime(caller).SendMessage(m)
 	}
 }
 
 func handleAPI(from *driver.Runtime, m *dipper.Message) {
 	s := Services[from.Service]
 	dipper.DeserializePayload(m)
-	resp := s.ResponseFactory.NewResponse(s, s.getDriverRuntime("eventbus").Output, m)
+	resp := s.ResponseFactory.NewResponse(s, s.GetReceiver("eventbus").(dipper.MessageReceiver), m)
 	if resp == nil {
 		dipper.Logger.Debugf("[%s] skipping handling API: %+v", s.name, m.Labels)
 
diff --git a/pkg/dipper/comm.go b/pkg/dipper/comm.go
index 69f8493..b226426 100644
--- a/pkg/dipper/comm.go
+++ b/pkg/dipper/comm.go
@@ -52,6 +52,23 @@ type Message struct {
 // MessageHandler : a type of functions that take a pointer to a message and handle it.
 type MessageHandler func(*Message)
 
+// MessageReceiver is a object that can receive and handle a dipper message.
+type MessageReceiver interface {
+	SendMessage(*Message)
+}
+
+// NullReceiver is a placeholder for MessageReceiver.
+type NullReceiver struct {
+	SendMessageFunc MessageHandler
+}
+
+// SendMessage does nothing unless overridden with SendMessageFunc.
+func (n *NullReceiver) SendMessage(msg *Message) {
+	if n.SendMessageFunc != nil {
+		n.SendMessageFunc(msg)
+	}
+}
+
 // SerializeContent : encode payload content into bytes.
 func SerializeContent(content interface{}) (ret []byte) {
 	var err error
diff --git a/pkg/dipper/driver.go b/pkg/dipper/driver.go
index b513dd6..b32f06e 100644
--- a/pkg/dipper/driver.go
+++ b/pkg/dipper/driver.go
@@ -198,9 +198,9 @@ func (d *Driver) GetLogger() *logging.Logger {
 	return Logger
 }
 
-// GetStream getting a output stream for a feature.
-func (d *Driver) GetStream(feature string) io.Writer {
-	return d.Out
+// GetReceiver getting a object that receive objects through SendMessage.
+func (d *Driver) GetReceiver(feature string) interface{} {
+	return d
 }
 
 // GetName returns the name of the driver.
diff --git a/pkg/dipper/mock_dipper/rpc.go b/pkg/dipper/mock_dipper/rpc.go
index c5d75d9..cce8baf 100644
--- a/pkg/dipper/mock_dipper/rpc.go
+++ b/pkg/dipper/mock_dipper/rpc.go
@@ -12,7 +12,6 @@
 package mock_dipper
 
 import (
-	io "io"
 	reflect "reflect"
 
 	gomock "github.com/golang/mock/gomock"
@@ -55,18 +54,18 @@ func (mr *MockRPCCallerStubMockRecorder) GetName() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetName", reflect.TypeOf((*MockRPCCallerStub)(nil).GetName))
 }
 
-// GetStream mocks base method.
-func (m *MockRPCCallerStub) GetStream(feature string) io.Writer {
+// GetReceiver mocks base method.
+func (m *MockRPCCallerStub) GetReceiver(feature string) interface{} {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetStream", feature)
-	ret0, _ := ret[0].(io.Writer)
+	ret := m.ctrl.Call(m, "GetReceiver", feature)
+	ret0, _ := ret[0].(interface{})
 	return ret0
 }
 
-// GetStream indicates an expected call of GetStream.
-func (mr *MockRPCCallerStubMockRecorder) GetStream(feature interface{}) *gomock.Call {
+// GetReceiver indicates an expected call of GetReceiver.
+func (mr *MockRPCCallerStubMockRecorder) GetReceiver(feature interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStream", reflect.TypeOf((*MockRPCCallerStub)(nil).GetStream), feature)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetReceiver", reflect.TypeOf((*MockRPCCallerStub)(nil).GetReceiver), feature)
 }
 
 // MockRPCCaller is a mock of RPCCaller interface.
diff --git a/pkg/dipper/rpc.go b/pkg/dipper/rpc.go
index 4d9652f..da50407 100644
--- a/pkg/dipper/rpc.go
+++ b/pkg/dipper/rpc.go
@@ -31,7 +31,7 @@ type RPCHandler func(string, string, []byte)
 // RPCCallerStub is an interface which every RPC caller should implement.
 type RPCCallerStub interface {
 	GetName() string
-	GetStream(feature string) io.Writer
+	GetReceiver(feature string) interface{}
 }
 
 // RPCCaller defines all method required for making rpc alls.
@@ -118,13 +118,13 @@ func (c *RPCCallerBase) CallRawNoWait(feature string, method string, params []by
 		rpcID = "skip"
 	}
 
-	out := c.Parent.GetStream(feature)
-	if out == nil {
+	receiver := c.Parent.GetReceiver(feature).(MessageReceiver)
+	if receiver == nil {
 		return fmt.Errorf("%w: feature not available: %s", ErrRPCError, feature)
 	}
 
 	// making the call by sending a message
-	SendMessage(out, &Message{
+	receiver.SendMessage(&Message{
 		Channel: c.Channel,
 		Subject: c.Subject,
 		Labels: map[string]string{


Test patch:
diff --git a/cmd/honeydipper/integration_test.go b/cmd/honeydipper/integration_test.go
index f345d5e..0d12c3f 100644
--- a/cmd/honeydipper/integration_test.go
+++ b/cmd/honeydipper/integration_test.go
@@ -200,7 +200,7 @@ func intTestDaemonShutdown(t *testing.T) {
 }
 
 func intTestDriverCrash(t *testing.T) {
-	assert.NotPanics(t, func() { _ = service.Services["operator"].GetStream("driver:gcloud-dataflow") }, "driver:gcloud-dataflow should be ready by now")
+	assert.NotPanics(t, func() { _ = service.Services["operator"].GetReceiver("driver:gcloud-dataflow") }, "driver:gcloud-dataflow should be ready by now")
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 	var (
diff --git a/internal/api/response_test.go b/internal/api/response_test.go
index d436b4e..29b745f 100644
--- a/internal/api/response_test.go
+++ b/internal/api/response_test.go
@@ -11,7 +11,6 @@ package api
 
 import (
 	"fmt"
-	"io"
 	"testing"
 	"time"
 
@@ -36,20 +35,17 @@ func responseTest(t *testing.T, c *ResponseTestCase) {
 	defer ctrl.Finish()
 	mockRPCCaller := mock_dipper.NewMockRPCCaller(ctrl)
 
-	eventbusO, eventbusI := io.Pipe()
-	defer eventbusO.Close()
-	defer eventbusI.Close()
+	stream := make(chan *dipper.Message, 1)
+	mockReceiver := &dipper.NullReceiver{
+		SendMessageFunc: func(msg *dipper.Message) {
+			stream <- msg
+		},
+	}
+
 	waitForMsg := func(delay time.Duration) *dipper.Message {
 		var m *dipper.Message
-		msgAvailable := make(chan byte)
-		go func() {
-			defer recover()
-			defer close(msgAvailable)
-			m = dipper.FetchMessage(eventbusO)
-			m.Size = 0 // ignore Size property, only used for raw msg
-		}()
 		select {
-		case <-msgAvailable:
+		case m = <-stream:
 		case <-time.After(delay):
 		}
 
@@ -67,7 +63,7 @@ func responseTest(t *testing.T, c *ResponseTestCase) {
 	factory := NewResponseFactory()
 	factory.DefsByName = c.defsByName
 
-	resp := factory.NewResponse(mockRPCCaller, eventbusI, c.msg)
+	resp := factory.NewResponse(mockRPCCaller, mockReceiver, c.msg)
 	if c.noResponse {
 		assert.Nil(t, resp)
 	} else {
diff --git a/internal/driver/builtin_test.go b/internal/driver/builtin_test.go
index ce2bbf6..9f418fb 100644
--- a/internal/driver/builtin_test.go
+++ b/internal/driver/builtin_test.go
@@ -13,6 +13,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/honeydipper/honeydipper/pkg/dipper"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -113,7 +114,8 @@ func TestBuiltinPrepare(t *testing.T) {
 					}
 				}
 			}()
-			c[0].(*BuiltinDriver).Prepare()
+			stream := make(chan *dipper.Message, 1)
+			c[0].(*BuiltinDriver).Prepare(stream)
 		}(tc.([]interface{}))
 	}
 }
diff --git a/internal/driver/driver_test.go b/internal/driver/driver_test.go
index d7ceb95..933cb10 100644
--- a/internal/driver/driver_test.go
+++ b/internal/driver/driver_test.go
@@ -53,7 +53,7 @@ func TestDriverNewDriver(t *testing.T) {
 					}
 				}
 			}()
-			NewDriver(c[0].(map[string]interface{}))
+			NewDriver("testfeature", c[0].(map[string]interface{}), nil, nil)
 		}(tc.([]interface{}))
 	}
 }
diff --git a/internal/service/service_test.go b/internal/service/service_test.go
index 70dbe1b..90c8681 100644
--- a/internal/service/service_test.go
+++ b/internal/service/service_test.go
@@ -29,18 +29,15 @@ func TestServiceLoopCatchError(t *testing.T) {
 		dipper.GetLogger("test service", "DEBUG", f, f)
 	}
 
+	stream := make(chan *dipper.Message, 1)
 	svc := &Service{
 		name: "testsvc",
 		driverRuntimes: map[string]*driver.Runtime{
 			"d1": {
-				State: driver.DriverAlive,
-				Handler: driver.NewDriver(map[string]interface{}{
-					"name": "testdriver1",
-					"type": "builtin",
-					"handlerData": map[string]interface{}{
-						"shortName": "testdriver1",
-					},
-				}),
+				Feature: "d1",
+				Handler: &driver.NullDriverHandler{},
+				State:   driver.DriverAlive,
+				Stream:  stream,
 			},
 		},
 		responders: map[string][]MessageResponder{
@@ -66,85 +63,77 @@ func TestServiceLoopCatchError(t *testing.T) {
 		},
 	}
 
-	svc.driverRuntimes["d1"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["d1"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND, 0o777)
 	go func() {
 		assert.NotPanics(t, svc.serviceLoop, "service loop should recover panic in route")
 	}()
 	// injecting error in route
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "error0",
 	}
 	time.Sleep(30 * time.Millisecond)
 	// quiting faster by send an extra message
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "noerror",
 	}
 	daemon.ShutDown()
-	daemon.ShuttingDown = false
 
-	svc.driverRuntimes["d1"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["d1"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND, 0o777)
+	daemon.ShuttingDown = false
+	stream = make(chan *dipper.Message, 1)
+	svc.driverRuntimes["d1"].Stream = stream
 	go func() {
 		assert.NotPanics(t, svc.serviceLoop, "service loop should recover panic in responder")
 	}()
 	// injecting error in responder
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "error1",
 	}
 	time.Sleep(30 * time.Millisecond)
 	// quiting faster by send an extra message
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "noerror",
 	}
 	daemon.ShutDown()
-	daemon.ShuttingDown = false
 
-	svc.driverRuntimes["d1"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["d1"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND, 0o777)
+	daemon.ShuttingDown = false
+	stream = make(chan *dipper.Message, 1)
+	svc.driverRuntimes["d1"].Stream = stream
 	go func() {
 		assert.NotPanics(t, svc.serviceLoop, "service loop should recover panic in transformer")
 	}()
 	// injecting error in transformer
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "error2",
 	}
 	time.Sleep(30 * time.Millisecond)
 	// quiting faster by send an extra message
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "noerror",
 	}
 	daemon.ShutDown()
-	daemon.ShuttingDown = false
 
-	svc.driverRuntimes["d1"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["d1"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND, 0o777)
+	daemon.ShuttingDown = false
+	stream = make(chan *dipper.Message, 1)
+	svc.driverRuntimes["d1"].Stream = stream
 	// injecting error in process
 	svc.driverRuntimes["d1"].Handler = nil
 	go func() {
 		assert.NotPanics(t, svc.serviceLoop, "service loop should recover panic in process itself")
 	}()
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "error3",
 	}
 	// recover the service object to avoid crash during quiting
-	svc.driverRuntimes["d1"].Handler = driver.NewDriver(map[string]interface{}{
-		"name": "testdriver1",
-		"type": "builtin",
-		"handlerData": map[string]interface{}{
-			"shortName": "testdriver1",
-		},
-	})
+	svc.driverRuntimes["d1"].Handler = &driver.NullDriverHandler{}
 	time.Sleep(30 * time.Millisecond)
 	// quiting faster by send an extra message
-	svc.driverRuntimes["d1"].Stream <- dipper.Message{
+	stream <- &dipper.Message{
 		Channel: "test",
 		Subject: "noerror",
 	}
@@ -159,29 +148,27 @@ func TestServiceRemoveEmitter(t *testing.T) {
 		dipper.GetLogger("test service", "DEBUG", f, f)
 	}
 
+	stream := make(chan *dipper.Message, 1)
 	svc := &Service{
 		name: "testsvc",
 		driverRuntimes: map[string]*driver.Runtime{
 			"driver:d1": {
-				State: driver.DriverAlive,
-				Handler: driver.NewDriver(map[string]interface{}{
-					"name": "d1",
-					"type": "builtin",
-					"handlerData": map[string]interface{}{
-						"shortName": "testdriver1",
-					},
+				Feature: "driver:d1",
+				Handler: driver.NewNullDriver(&driver.Meta{
+					Name: "d1",
+					Type: "null",
 				}),
+				State:  driver.DriverAlive,
+				Stream: stream,
 			},
 			"emitter": {
-				State: driver.DriverAlive,
-				Handler: driver.NewDriver(map[string]interface{}{
-					"name": "test-emitter",
-					"type": "builtin",
-					"handlerData": map[string]interface{}{
-						"shortName": "testdriver1",
-					},
-				}),
 				Feature: "emitter",
+				Handler: driver.NewNullDriver(&driver.Meta{
+					Name: "test-emitter",
+					Type: "null",
+				}),
+				State:  driver.DriverAlive,
+				Stream: make(chan *dipper.Message),
 			},
 		},
 		Route: func(m *dipper.Message) []RoutedMessage {
@@ -191,10 +178,6 @@ func TestServiceRemoveEmitter(t *testing.T) {
 	daemon.Emitters["testsvc"] = svc
 
 	daemon.ShuttingDown = false
-	svc.driverRuntimes["driver:d1"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["driver:d1"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND, 0o777)
-	svc.driverRuntimes["emitter"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["emitter"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND|os.O_WRONLY, 0o777)
 	go func() {
 		assert.NotPanics(t, svc.serviceLoop, "service loop should recover panic if emitter is removed")
 	}()
@@ -206,7 +189,7 @@ func TestServiceRemoveEmitter(t *testing.T) {
 		assert.NotPanics(t, func() {
 			for i := 0; i < 50; i++ {
 				select {
-				case svc.driverRuntimes["driver:d1"].Stream <- dipper.Message{
+				case stream <- &dipper.Message{
 					Channel: "test",
 					Subject: "noerror",
 				}:
@@ -234,7 +217,7 @@ func TestServiceRemoveEmitter(t *testing.T) {
 					"drivers": map[string]interface{}{
 						"d1": map[string]interface{}{
 							"name": "d1",
-							"type": "builtin",
+							"type": "null",
 						},
 					},
 				},
@@ -258,29 +241,28 @@ func TestServiceEmitterCrashing(t *testing.T) {
 		dipper.GetLogger("test service", "DEBUG", f, f)
 	}
 
+	stream := make(chan *dipper.Message, 1)
+	emitterStream := make(chan *dipper.Message, 1)
 	svc := &Service{
 		name: "testsvc",
 		driverRuntimes: map[string]*driver.Runtime{
 			"driver:d1": {
-				State: driver.DriverAlive,
-				Handler: driver.NewDriver(map[string]interface{}{
-					"name": "d1",
-					"type": "builtin",
-					"handlerData": map[string]interface{}{
-						"shortName": "testdriver1",
-					},
-				}),
+				Feature: "d1",
+				Handler: &driver.NullDriverHandler{},
+				State:   driver.DriverAlive,
+				Stream:  stream,
 			},
 			"emitter": {
-				State: driver.DriverAlive,
-				Handler: driver.NewDriver(map[string]interface{}{
-					"name": "test-emitter",
-					"type": "builtin",
-					"handlerData": map[string]interface{}{
-						"shortName": "testdriver1",
+				Feature: "test-emitter",
+				Handler: &driver.NullDriverHandler{
+					SendMessageFunc: func(*dipper.Message) {
+						if emitterStream == nil {
+							panic("emitter crashed")
+						}
 					},
-				}),
-				Feature: "emitter",
+				},
+				State:  driver.DriverAlive,
+				Stream: emitterStream,
 			},
 		},
 		Route: func(m *dipper.Message) []RoutedMessage {
@@ -290,10 +272,6 @@ func TestServiceEmitterCrashing(t *testing.T) {
 	daemon.Emitters["testsvc"] = svc
 
 	daemon.ShuttingDown = false
-	svc.driverRuntimes["driver:d1"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["driver:d1"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND, 0o777)
-	svc.driverRuntimes["emitter"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["emitter"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND|os.O_WRONLY, 0o777)
 	go func() {
 		assert.NotPanics(t, svc.serviceLoop, "service loop should recover panic if emitter crashes")
 	}()
@@ -305,7 +283,7 @@ func TestServiceEmitterCrashing(t *testing.T) {
 		assert.NotPanics(t, func() {
 			for i := 0; i < 50; i++ {
 				select {
-				case svc.driverRuntimes["driver:d1"].Stream <- dipper.Message{
+				case stream <- &dipper.Message{
 					Channel: "test",
 					Subject: "noerror",
 				}:
@@ -321,8 +299,8 @@ func TestServiceEmitterCrashing(t *testing.T) {
 	// mark it as failed to avoid restarting the driver
 	svc.driverRuntimes["emitter"].State = driver.DriverFailed
 	// crash emitter
-	svc.driverRuntimes["emitter"].Output.Close()
-	close(svc.driverRuntimes["emitter"].Stream)
+	close(emitterStream)
+	emitterStream = nil
 	time.Sleep(100 * time.Millisecond)
 
 	daemon.ShutDown()
@@ -335,29 +313,31 @@ func TestServiceReplaceEmitter(t *testing.T) {
 		dipper.GetLogger("test service", "DEBUG", f, f)
 	}
 
+	stream := make(chan *dipper.Message, 1)
+	emitterStream := make(chan *dipper.Message, 1)
 	svc := &Service{
 		name: "testsvc",
 		driverRuntimes: map[string]*driver.Runtime{
 			"driver:d1": {
-				State: driver.DriverAlive,
-				Handler: driver.NewDriver(map[string]interface{}{
-					"name": "d1",
-					"type": "builtin",
-					"handlerData": map[string]interface{}{
-						"shortName": "testdriver1",
-					},
+				Feature: "driver:d1",
+				Handler: driver.NewNullDriver(&driver.Meta{
+					Name: "d1",
+					Type: "null",
 				}),
+				State:  driver.DriverAlive,
+				Stream: stream,
 			},
 			"emitter": {
-				State: driver.DriverAlive,
-				Handler: driver.NewDriver(map[string]interface{}{
-					"name": "test-emitter",
-					"type": "builtin",
-					"handlerData": map[string]interface{}{
-						"shortName": "testdriver1",
+				Feature: "test-emitter",
+				Handler: &driver.NullDriverHandler{
+					SendMessageFunc: func(*dipper.Message) {
+						if emitterStream == nil {
+							panic("emitter crashed")
+						}
 					},
-				}),
-				Feature: "emitter",
+				},
+				State:  driver.DriverAlive,
+				Stream: emitterStream,
 			},
 		},
 		Route: func(m *dipper.Message) []RoutedMessage {
@@ -367,10 +347,6 @@ func TestServiceReplaceEmitter(t *testing.T) {
 	daemon.Emitters["testsvc"] = svc
 
 	daemon.ShuttingDown = false
-	svc.driverRuntimes["driver:d1"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["driver:d1"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND, 0o777)
-	svc.driverRuntimes["emitter"].Stream = make(chan dipper.Message, 1)
-	svc.driverRuntimes["emitter"].Output, _ = os.OpenFile(os.DevNull, os.O_APPEND|os.O_WRONLY, 0o777)
 	go func() {
 		assert.NotPanics(t, svc.serviceLoop, "service loop should recover panic if emitter is changed")
 	}()
@@ -382,7 +358,7 @@ func TestServiceReplaceEmitter(t *testing.T) {
 		assert.NotPanics(t, func() {
 			for i := 0; i < 50; i++ {
 				select {
-				case svc.driverRuntimes["driver:d1"].Stream <- dipper.Message{
+				case stream <- &dipper.Message{
 					Channel: "test",
 					Subject: "noerror",
 				}:
@@ -416,9 +392,11 @@ func TestServiceReplaceEmitter(t *testing.T) {
 					"drivers": map[string]interface{}{
 						"d1": map[string]interface{}{
 							"name": "d1",
+							"type": "null",
 						},
 						"emitter2": map[string]interface{}{
 							"name": "emitter2",
+							"type": "null",
 						},
 					},
 				},
diff --git a/pkg/dipper/encryption_test.go b/pkg/dipper/encryption_test.go
index c982153..cdda32b 100644
--- a/pkg/dipper/encryption_test.go
+++ b/pkg/dipper/encryption_test.go
@@ -10,7 +10,6 @@
 package dipper
 
 import (
-	"io"
 	"testing"
 
 	"github.com/ghodss/yaml"
@@ -19,32 +18,26 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func wrapDecryptAll(t *testing.T, doc string, expect func(io.Reader, *RPCCallerBase)) map[string]interface{} {
+func wrapDecryptAll(t *testing.T, doc string, expect []MessageReceiver, c *RPCCallerBase) map[string]interface{} {
 	var data map[string]interface{}
 	assert.NoError(t, yaml.Unmarshal([]byte(doc), &data))
 
-	O, I := io.Pipe()
-
+	curr := 0
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	mockStub := mock_dipper.NewMockRPCCallerStub(ctrl)
-	c := &RPCCallerBase{}
 	mockStub.EXPECT().GetName().AnyTimes().Return("mockCaller")
-	mockStub.EXPECT().GetStream(gomock.Any()).AnyTimes().Return(I)
-	c.Init(mockStub, "rpc", "call")
+	mockStub.EXPECT().GetReceiver(gomock.AssignableToTypeOf("")).AnyTimes().DoAndReturn(func(x string) MessageReceiver {
+		ret := expect[curr]
+		curr++
 
-	done := make(chan struct{})
-	go func() {
-		assert.NotPanics(t, func() { expect(O, c) }, "should not panic when sending and receiving messages")
-		assert.PanicsWithError(t, "invalid message envelope: io: read/write on closed pipe", func() { FetchRawMessage(O) }, "should contain no more messages")
-		close(done)
-	}()
+		return ret
+	})
 
-	DecryptAll(c, data)
+	*c = RPCCallerBase{}
+	c.Init(mockStub, "rpc", "call")
 
-	O.Close()
-	I.Close()
-	<-done
+	DecryptAll(c, data)
 
 	return data
 }
@@ -56,41 +49,46 @@ data:
   item2: ENC[noexist,YWFiYmNjZGQ=]
 `
 
-	expect := func(O io.Reader, c *RPCCallerBase) {
-		assert.Equalf(t,
-			Message{
-				Channel: "rpc",
-				Subject: "call",
-				IsRaw:   true,
-				Size:    8,
-				Payload: []byte("aabbccdd"),
-				Labels: map[string]string{
-					"caller":  "-",
-					"feature": "driver:noexist",
-					"method":  "decrypt",
-					"rpcID":   "0",
-				},
-			},
-			*FetchRawMessage(O),
-			"should make call to decrypt item2",
-		)
-		c.HandleReturn(&Message{
-			Channel: "rpc",
-			Subject: "return",
-			IsRaw:   true,
-			Size:    18,
-			Payload: []byte("decrypted aabbccdd"),
-			Labels: map[string]string{
-				"caller":  "-",
-				"feature": "driver:noexist",
-				"method":  "decrypt",
-				"rpcID":   "0",
-				"status":  "success",
+	c := &RPCCallerBase{}
+
+	expect := []MessageReceiver{
+		&NullReceiver{
+			SendMessageFunc: func(msg *Message) {
+				assert.Equalf(t,
+					Message{
+						Channel: "rpc",
+						Subject: "call",
+						IsRaw:   true,
+						Payload: []byte("aabbccdd"),
+						Labels: map[string]string{
+							"caller":  "-",
+							"feature": "driver:noexist",
+							"method":  "decrypt",
+							"rpcID":   "0",
+						},
+					},
+					*msg,
+					"should make call to decrypt item2",
+				)
+				c.HandleReturn(&Message{
+					Channel: "rpc",
+					Subject: "return",
+					IsRaw:   true,
+					Size:    18,
+					Payload: []byte("decrypted aabbccdd"),
+					Labels: map[string]string{
+						"caller":  "-",
+						"feature": "driver:noexist",
+						"method":  "decrypt",
+						"rpcID":   "0",
+						"status":  "success",
+					},
+				})
 			},
-		})
+		},
 	}
 
-	data := wrapDecryptAll(t, doc, expect)
+	data := wrapDecryptAll(t, doc, expect, c)
 	assert.Equal(t, "decrypted aabbccdd", MustGetMapDataStr(data, "data.item2"), "data.item2 should container decrypted data")
 	assert.Equal(t, "not encrypted", MustGetMapDataStr(data, "data.item1"), "data.item1 should remain unchanged")
 }
@@ -103,10 +101,17 @@ data:
     item4: ENC[deferred,driver1,YWFiYmNjZGQ=]
 `
 
-	expect := func(O io.Reader, c *RPCCallerBase) {
+	c := &RPCCallerBase{}
+
+	expect := []MessageReceiver{
+		&NullReceiver{
+			SendMessageFunc: func(msg *Message) {
+				assert.Fail(t, "should not make calls for delayed decryption.")
+			},
+		},
 	}
 
-	data := wrapDecryptAll(t, doc, expect)
+	data := wrapDecryptAll(t, doc, expect, c)
 	assert.Equal(t, "ENC[driver1,YWFiYmNjZGQ=]", MustGetMapDataStr(data, "data.item3.item4"), "item4 should be stripped off one deferred flag")
 	assert.Equal(t, "not encrypted", MustGetMapDataStr(data, "data.item1"), "data.item1 should remain unchanged")
 }
@@ -118,41 +123,46 @@ data:
   item2: LOOKUP[kvstore,foo]
 `
 
-	expect := func(O io.Reader, c *RPCCallerBase) {
-		assert.Equalf(t,
-			Message{
-				Channel: "rpc",
-				Subject: "call",
-				IsRaw:   true,
-				Size:    3,
-				Payload: []byte("foo"),
-				Labels: map[string]string{
-					"caller":  "-",
-					"feature": "driver:kvstore",
-					"method":  "lookup",
-					"rpcID":   "0",
-				},
-			},
-			*FetchRawMessage(O),
-			"should make call to lookup for item2",
-		)
-		c.HandleReturn(&Message{
-			Channel: "rpc",
-			Subject: "return",
-			IsRaw:   true,
-			Size:    3,
-			Payload: []byte("bar"),
-			Labels: map[string]string{
-				"caller":  "-",
-				"feature": "driver:kvstore",
-				"method":  "lookup",
-				"rpcID":   "0",
-				"status":  "success",
+	c := &RPCCallerBase{}
+
+	expect := []MessageReceiver{
+		&NullReceiver{
+			SendMessageFunc: func(msg *Message) {
+				assert.Equalf(t,
+					Message{
+						Channel: "rpc",
+						Subject: "call",
+						IsRaw:   true,
+						Payload: []byte("foo"),
+						Labels: map[string]string{
+							"caller":  "-",
+							"feature": "driver:kvstore",
+							"method":  "lookup",
+							"rpcID":   "0",
+						},
+					},
+					*msg,
+					"should make call to lookup for item2",
+				)
+				c.HandleReturn(&Message{
+					Channel: "rpc",
+					Subject: "return",
+					IsRaw:   true,
+					Size:    3,
+					Payload: []byte("bar"),
+					Labels: map[string]string{
+						"caller":  "-",
+						"feature": "driver:kvstore",
+						"method":  "lookup",
+						"rpcID":   "0",
+						"status":  "success",
+					},
+				})
 			},
-		})
+		},
 	}
 
-	data := wrapDecryptAll(t, doc, expect)
+	data := wrapDecryptAll(t, doc, expect, c)
 	assert.Equal(t, "bar", MustGetMapDataStr(data, "data.item2"), "data.item2 should container decrypted data")
 	assert.Equal(t, "not encrypted", MustGetMapDataStr(data, "data.item1"), "data.item1 should remain unchanged")
 }
diff --git a/pkg/dipper/rpc_test.go b/pkg/dipper/rpc_test.go
index bab3f1f..957c11c 100644
--- a/pkg/dipper/rpc_test.go
+++ b/pkg/dipper/rpc_test.go
@@ -10,9 +10,6 @@
 package dipper
 
 import (
-	"bytes"
-	"fmt"
-	"io"
 	"testing"
 	"time"
 
@@ -25,72 +22,48 @@ func TestRPCCallRaw(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
-	mockStub := mock_dipper.NewMockRPCCallerStub(ctrl)
-
-	var b bytes.Buffer
+	curr := 0
 	c := RPCCallerBase{}
+	var m *Message
+
+	expect := []MessageReceiver{
+		&NullReceiver{
+			SendMessageFunc: func(msg *Message) {
+				m = msg
+			},
+		},
+	}
 
+	mockStub := mock_dipper.NewMockRPCCallerStub(ctrl)
 	mockStub.EXPECT().GetName().AnyTimes().Return("mockCaller")
-	mockStub.EXPECT().GetStream(gomock.Any()).Return(&b)
+	mockStub.EXPECT().GetReceiver(gomock.AssignableToTypeOf("")).AnyTimes().DoAndReturn(func(x string) MessageReceiver {
+		ret := expect[curr]
+		curr++
+
+		return ret
+	})
 	c.Init(mockStub, "rpc", "call")
 	assert.NotPanicsf(t, func() { c.CallRaw("target", "testmethod", []byte("hello world")) }, "CallRaw should not panic")
 	time.Sleep(time.Second / 10)
-	var channel, subject string
-	var size, numlabels int
-	fmt.Fscanln(&b, &channel, &subject, &numlabels, &size)
-	assert.Equal(t, "rpc", channel, "rpc call sends message through rpc channel")
-	assert.Equal(t, "call", subject, "rpc uses callee and method and prefix for subject")
-	assert.Equal(t, 11, size, "rpc call raw sends the bytes as payload")
-	assert.Equal(t, 4, numlabels, "rpc call use labels to specify feature and method")
-	var lname string
-	var lval []byte
-	var vl int
-	labels := map[string]string{}
-	fmt.Fscanln(&b, &lname, &vl)
-	if vl > 0 {
-		lval = make([]byte, vl)
-		Must(io.ReadFull(&b, lval))
-		labels[lname] = string(lval)
-	} else {
-		labels[lname] = ""
-	}
-	fmt.Fscanln(&b, &lname, &vl)
-	if vl > 0 {
-		lval = make([]byte, vl)
-		Must(io.ReadFull(&b, lval))
-		labels[lname] = string(lval)
-	} else {
-		labels[lname] = ""
-	}
-	fmt.Fscanln(&b, &lname, &vl)
-	if vl > 0 {
-		lval = make([]byte, vl)
-		Must(io.ReadFull(&b, lval))
-		labels[lname] = string(lval)
-	} else {
-		labels[lname] = ""
-	}
-	fmt.Fscanln(&b, &lname, &vl)
-	if vl > 0 {
-		lval = make([]byte, vl)
-		Must(io.ReadFull(&b, lval))
-		labels[lname] = string(lval)
-	} else {
-		labels[lname] = ""
-	}
-	lv, ok := labels["caller"]
+
+	assert.Equal(t, "rpc", m.Channel, "rpc call sends message through rpc channel")
+	assert.Equal(t, "call", m.Subject, "rpc uses callee and method and prefix for subject")
+	assert.Equal(t, 4, len(m.Labels), "rpc call use labels to specify feature and method")
+
+	lv, ok := m.Labels["caller"]
 	assert.True(t, ok, "rpc caller present")
 	assert.Equal(t, "-", lv, "rpc caller should be -")
-	lv, ok = labels["rpcID"]
+	lv, ok = m.Labels["rpcID"]
 	assert.True(t, ok, "rpc rpcID present")
 	assert.Equal(t, "0", lv, "rpcID should be 0")
-	lv, ok = labels["feature"]
+	lv, ok = m.Labels["feature"]
 	assert.True(t, ok, "rpc feature present")
 	assert.Equal(t, "target", lv, "feature should be target")
-	lv, ok = labels["method"]
+	lv, ok = m.Labels["method"]
 	assert.True(t, ok, "rpc method present")
+
 	assert.Equal(t, "testmethod", lv, "method should be testmethod")
-	received, err := io.ReadAll(&b)
-	assert.Nil(t, err, "rpc call payload should be readable")
+	received, ok := m.Payload.([]byte)
+	assert.True(t, ok, "rpc call payload should be byte array")
 	assert.Equal(t, "hello world", string(received), "rpc should be unchanged")
 }