[plc4x] branch next-gen-core updated: added send context to conversation context

[sr...@apache.org]

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

sruehl pushed a commit to branch next-gen-core
in repository https://gitbox.apache.org/repos/asf/plc4x.git


The following commit(s) were added to refs/heads/next-gen-core by this push:
     new f5d9ab9  added send context to conversation context
f5d9ab9 is described below

commit f5d9ab953765a2874958eeccc81c96952a258dc5
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Dec 18 16:21:59 2019 +0100

    added send context to conversation context
---
 .../apache/plc4x/java/spi/ConversationContext.java | 48 ++++++++++++++++++++++
 .../apache/plc4x/java/spi/Plc4xNettyWrapper.java   | 21 +++++++---
 2 files changed, 64 insertions(+), 5 deletions(-)

diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/ConversationContext.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/ConversationContext.java
index 973d9a3..7c6104a 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/ConversationContext.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/ConversationContext.java
@@ -19,10 +19,58 @@
 
 package org.apache.plc4x.java.spi;
 
+import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
+
+import java.time.Duration;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
 public interface ConversationContext<T> {
 
     void sendToWire(T msg);
 
     void fireConnected();
 
+    SendRequestContext<T> sendRequest(T packet);
+
+    interface SendRequestContext<T> {
+
+        SendRequestContext<T> expectResponse(Class<T> clazz, Duration timeout);
+
+        SendRequestContext<T> check(Function<T, Boolean> checker);
+
+        SendRequestContext<T> handle(Consumer<T> packetConsumer);
+
+        <E extends Throwable> SendRequestContext<T> onTimeout(BiConsumer<T, E> packetConsumer);
+
+        <E extends Throwable> SendRequestContext<T> onError(BiConsumer<T, E> packetConsumer);
+
+        SendRequestContext<T> onSuccess(Consumer<T> packetConsumer);
+
+        <R> SendRequestContext<R> unwrap(Function<T, R> unwrapper);
+
+        <R> SendRequestContext<R> unwrap(Class<R> clazz, Function<T, R> unwrapper);
+
+        void finish();
+    }
+
+    class PlcCompletionException extends PlcRuntimeException {
+
+        public PlcCompletionException(String message) {
+            super(message);
+        }
+
+        public PlcCompletionException(String message, Throwable cause) {
+            super(message, cause);
+        }
+
+        public PlcCompletionException(Throwable cause) {
+            super(cause);
+        }
+
+        public PlcCompletionException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+            super(message, cause, enableSuppression, writableStackTrace);
+        }
+    }
 }
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/Plc4xNettyWrapper.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/Plc4xNettyWrapper.java
index e9e60e2..a0b83f1 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/Plc4xNettyWrapper.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/Plc4xNettyWrapper.java
@@ -40,17 +40,20 @@ public class Plc4xNettyWrapper<T> extends MessageToMessageCodec<T, PlcRequestCon
         this.protocolBase = parent;
     }
 
-    @Override protected void encode(ChannelHandlerContext channelHandlerContext, PlcRequestContainer plcRequestContainer, List<Object> list) throws Exception {
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, PlcRequestContainer plcRequestContainer, List<Object> list) throws Exception {
         logger.info("Encoding {}", plcRequestContainer);
         protocolBase.encode(new DefaultConversationContext<T>(channelHandlerContext) {
-            @Override public void sendToWire(T msg) {
+            @Override
+            public void sendToWire(T msg) {
                 logger.info("Sending to wire {}", msg);
                 list.add(msg);
             }
         }, plcRequestContainer);
     }
 
-    @Override protected void decode(ChannelHandlerContext channelHandlerContext, T t, List<Object> list) throws Exception {
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, T t, List<Object> list) throws Exception {
         logger.info("Decoding {}", t);
         protocolBase.decode(new DefaultConversationContext<>(channelHandlerContext), t);
     }
@@ -73,14 +76,22 @@ public class Plc4xNettyWrapper<T> extends MessageToMessageCodec<T, PlcRequestCon
             this.channelHandlerContext = channelHandlerContext;
         }
 
-        @Override public void sendToWire(T msg) {
+        @Override
+        public void sendToWire(T msg) {
             logger.info("Sending to wire {}", msg);
             channelHandlerContext.channel().writeAndFlush(msg);
         }
 
-        @Override public void fireConnected() {
+        @Override
+        public void fireConnected() {
             logger.info("Firing Connected!");
             channelHandlerContext.pipeline().fireUserEventTriggered(new ConnectedEvent());
         }
+
+        @Override
+        public SendRequestContext<T> sendRequest(T packet) {
+            // TODO: implement me
+            return null;
+        }
     }
 }