[servicecomb-pack] 05/21: SCB-1418 add KafkaActorEventChannel.kafkaMessagePublisher field

[ni...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git

commit 35380b8ae96bcf00c834d5f773ec9d86ba53fbbd
Author: CMonkey <42...@gmail.com>
AuthorDate: Tue Aug 13 11:47:18 2019 +0800

    SCB-1418 add KafkaActorEventChannel.kafkaMessagePublisher field
---
 .../pack/alpha/fsm/channel/KafkaActorEventChannel.java            | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/KafkaActorEventChannel.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/KafkaActorEventChannel.java
index 331d212..aca2676 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/KafkaActorEventChannel.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/KafkaActorEventChannel.java
@@ -19,6 +19,7 @@ package org.apache.servicecomb.pack.alpha.fsm.channel;
 
 import java.lang.invoke.MethodHandles;
 import org.apache.servicecomb.pack.alpha.core.fsm.event.base.BaseEvent;
+import org.apache.servicecomb.pack.alpha.fsm.channel.kafka.KafkaMessagePublisher;
 import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService;
 import org.apache.servicecomb.pack.alpha.core.fsm.sink.ActorEventSink;
 import org.slf4j.Logger;
@@ -27,13 +28,16 @@ import org.slf4j.LoggerFactory;
 public class KafkaActorEventChannel extends AbstractActorEventChannel {
   private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
+  private KafkaMessagePublisher kafkaMessagePublisher;
+
   public KafkaActorEventChannel(
-      ActorEventSink actorEventSink, MetricsService metricsService) {
+      ActorEventSink actorEventSink, MetricsService metricsService, KafkaMessagePublisher kafkaMessagePublisher) {
     super(actorEventSink, metricsService);
+    this.kafkaMessagePublisher = kafkaMessagePublisher;
   }
 
   @Override
   public void sendTo(BaseEvent event){
-    throw new UnsupportedOperationException("Doesn't implement yet!");
+      kafkaMessagePublisher.publish(event);
   }
 }