1 files changed, 27 insertions, 22 deletions

diff --git a/yardstick/benchmark/runners/base.py b/yardstick/benchmark/runners/base.py
index af2557441..94de45d1e 100755
--- a/yardstick/benchmark/runners/base.py
+++ b/yardstick/benchmark/runners/base.py
@@ -25,9 +25,6 @@ import traceback
 from six import moves
 
 from yardstick.benchmark.scenarios import base as base_scenario
-from yardstick.common import messaging
-from yardstick.common.messaging import payloads
-from yardstick.common.messaging import producer
 from yardstick.common import utils
 from yardstick.dispatcher.base import Base as DispatcherBase
 
@@ -80,6 +77,33 @@ def _periodic_action(interval, command, queue):
         queue.put({'periodic-action-data': data})
 
 
+class ScenarioOutput(dict):
+
+    QUEUE_PUT_TIMEOUT = 10
+
+    def __init__(self, queue, **kwargs):
+        super(ScenarioOutput, self).__init__()
+        self._queue = queue
+        self.result_ext = dict()
+        for key, val in kwargs.items():
+            self.result_ext[key] = val
+            setattr(self, key, val)
+
+    def push(self, data=None, add_timestamp=True):
+        if data is None:
+            data = dict(self)
+
+        if add_timestamp:
+            result = {'timestamp': time.time(), 'data': data}
+        else:
+            result = data
+
+        for key in self.result_ext.keys():
+            result[key] = getattr(self, key)
+
+        self._queue.put(result, True, self.QUEUE_PUT_TIMEOUT)
+
+
 class Runner(object):
     runners = []
 
@@ -271,22 +295,3 @@ class Runner(object):
         dispatchers = DispatcherBase.get(self.config['output_config'])
         dispatcher = next((d for d in dispatchers if d.__dispatcher_type__ == 'Influxdb'))
         dispatcher.upload_one_record(record, self.case_name, '', task_id=self.task_id)
-
-
-class RunnerProducer(producer.MessagingProducer):
-    """Class implementing the message producer for runners"""
-
-    def __init__(self, _id):
-        super(RunnerProducer, self).__init__(messaging.TOPIC_RUNNER, _id=_id)
-
-    def start_iteration(self, version=1, data=None):
-        data = {} if not data else data
-        self.send_message(
-            messaging.RUNNER_METHOD_START_ITERATION,
-            payloads.RunnerPayload(version=version, data=data))
-
-    def stop_iteration(self, version=1, data=None):
-        data = {} if not data else data
-        self.send_message(
-            messaging.RUNNER_METHOD_STOP_ITERATION,
-            payloads.RunnerPayload(version=version, data=data))