diff --git a/integrations/apache-edgent/src/main/java/org/apache/plc4x/edgent/PlcConnectionAdapter.java b/integrations/apache-edgent/src/main/java/org/apache/plc4x/edgent/PlcConnectionAdapter.java
index 2bff661..ab55af8 100644
--- a/integrations/apache-edgent/src/main/java/org/apache/plc4x/edgent/PlcConnectionAdapter.java
+++ b/integrations/apache-edgent/src/main/java/org/apache/plc4x/edgent/PlcConnectionAdapter.java
@@ -18,8 +18,7 @@
 */
 package org.apache.plc4x.edgent;
 
-import java.util.Calendar;
-
+import com.google.gson.JsonObject;
 import org.apache.edgent.function.Consumer;
 import org.apache.edgent.function.Function;
 import org.apache.edgent.function.Supplier;
@@ -28,18 +27,19 @@
 import org.apache.plc4x.java.api.connection.PlcReader;
 import org.apache.plc4x.java.api.connection.PlcWriter;
 import org.apache.plc4x.java.api.exceptions.PlcException;
-import org.apache.plc4x.java.api.messages.*;
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcWriteRequest;
 import org.apache.plc4x.java.api.model.Address;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.gson.JsonObject;
+import java.util.Calendar;
 
 /**
  * PlcConnectionAdapter encapsulates a plc4x {@link PlcConnection}.
  * <p>
  * The idea here is to use PlcConnectionAdapter to enable our Edgent Supplier/Consumer
- * instances to be isolated from the details of / variability of 
+ * instances to be isolated from the details of / variability of
  * PlcConnection mgmt and such.
  * <p>
  * A PlcConnectionAdapter is subject to the constraints of the underlying
@@ -53,17 +53,17 @@
  * support for multiple connections from a single client.
  * <p>
  * A single PlcConnectionAdapter can be used by multiple threads concurrently
- * (e.g., used by multiple PlcFunctions Consumers for {@code Topology.poll()} and/or 
- * multiple Suppliers for {@code TStream.sink()}). 
- * 
+ * (e.g., used by multiple PlcFunctions Consumers for {@code Topology.poll()} and/or
+ * multiple Suppliers for {@code TStream.sink()}).
+ *
  * @see PlcFunctions
  */
-public class PlcConnectionAdapter implements AutoCloseable{
+public class PlcConnectionAdapter implements AutoCloseable {
 
-  private static final Logger logger = LoggerFactory.getLogger(PlcConnectionAdapter.class);
-  
-  private String plcConnectionUrl;
-  private PlcConnection plcConnection;
+    private static final Logger logger = LoggerFactory.getLogger(PlcConnectionAdapter.class);
+
+    private String plcConnectionUrl;
+    private PlcConnection plcConnection;
   
   /*
    * NOTES:
@@ -71,130 +71,127 @@
    *   of read or write errors, my thinking is to enhance the PlcConnectionAdapter
    *   to enable the app to register an error callback handler or such.
    */
-  
-  public PlcConnectionAdapter(PlcConnection plcConnection) {
-    this.plcConnection = plcConnection;
-  }
-  
-  public PlcConnectionAdapter(String plcConnectionUrl) {
-    this.plcConnectionUrl = plcConnectionUrl;
-  }
-  
-  PlcConnection getConnection() throws PlcException {
-    synchronized(this) {
-      if (plcConnection == null) {
-        plcConnection = new PlcDriverManager().getConnection(plcConnectionUrl);
-      }
-      return plcConnection;
+
+    public PlcConnectionAdapter(PlcConnection plcConnection) {
+        this.plcConnection = plcConnection;
     }
-  }
 
-  @Override
-  public void close() throws Exception {
-    // only close a connection this instance created/connected
-    if (plcConnectionUrl != null) {
-      if (plcConnection != null)
-        plcConnection.close();
+    public PlcConnectionAdapter(String plcConnectionUrl) {
+        this.plcConnectionUrl = plcConnectionUrl;
     }
-  }
 
-  <T> Supplier<T> newSupplier(Class<T> datatype, String addressStr) {
-    PlcConnectionAdapter.checkDatatype(datatype);
-    return new Supplier<T>() {
-      private static final long serialVersionUID = 1L;
-
-      @Override
-      public T get() {
-        PlcConnection connection = null;
-        Address address = null;
-        try {
-          connection = getConnection();
-          address = connection.parseAddress(addressStr);
-          PlcReader reader = connection.getReader().get();
-          PlcReadRequest readRequest = PlcConnectionAdapter.newPlcReadRequest(datatype, address);
-          T value = (T) reader.read(readRequest).get().getResponseItems().get(0).getValues().get(0);
-          return value;
+    PlcConnection getConnection() throws PlcException {
+        synchronized (this) {
+            if (plcConnection == null) {
+                plcConnection = new PlcDriverManager().getConnection(plcConnectionUrl);
+            }
+            return plcConnection;
         }
-        catch (Exception e) {
-          logger.error("reading from plc device {} {} failed", connection, address, e);
-          return null;
+    }
+
+    @Override
+    public void close() throws Exception {
+        // only close a connection this instance created/connected
+        if (plcConnectionUrl != null) {
+            if (plcConnection != null)
+                plcConnection.close();
         }
-      }
-      
-    };
-  }
+    }
 
-  <T> Consumer<T> newConsumer(Class<T> datatype, String addressStr) {
-    PlcConnectionAdapter.checkDatatype(datatype);
-    return new Consumer<T>() {
-      private static final long serialVersionUID = 1L;
+    <T> Supplier<T> newSupplier(Class<T> datatype, String addressStr) {
+        PlcConnectionAdapter.checkDatatype(datatype);
+        return new Supplier<T>() {
+            private static final long serialVersionUID = 1L;
 
-      @Override
-      public void accept(T arg0) {
-        PlcConnection connection = null;
-        Address address = null;
-        try {
-          connection = getConnection();
-          address = connection.parseAddress(addressStr);
-          PlcWriter writer = connection.getWriter().get();
-          PlcWriteRequest writeReq = PlcConnectionAdapter.newPlcWriteRequest(address, arg0);
-          writer.write(writeReq).get();
-        }
-        catch (Exception e) {
-          logger.error("writing to plc device {} {} failed", connection, address, e);
-        }
-      }
-      
-    };
-  }
-  
-  <T> Consumer<JsonObject> newConsumer(Class<T> datatype, Function<JsonObject,String> addressFn, Function<JsonObject,T> valueFn) {
-    PlcConnectionAdapter.checkDatatype(datatype);
-    return new Consumer<JsonObject>() {
-      private static final long serialVersionUID = 1L;
+            @Override
+            public T get() {
+                PlcConnection connection = null;
+                Address address = null;
+                try {
+                    connection = getConnection();
+                    address = connection.parseAddress(addressStr);
+                    PlcReader reader = connection.getReader().get();
+                    PlcReadRequest<T> readRequest = PlcConnectionAdapter.newPlcReadRequest(datatype, address);
+                    T value = (T) reader.read(readRequest).get().getResponseItems().get(0).getValues().get(0);
+                    return value;
+                } catch (Exception e) {
+                    logger.error("reading from plc device {} {} failed", connection, address, e);
+                    return null;
+                }
+            }
 
-      @Override
-      public void accept(JsonObject jo) {
-        PlcConnection connection = null;
-        Address address = null;
-        try {
-          connection = getConnection();
-          String addressStr = addressFn.apply(jo);
-          address = connection.parseAddress(addressStr);
-          T value = valueFn.apply(jo);
-          PlcWriter writer = connection.getWriter().get();
-          PlcWriteRequest writeReq = newPlcWriteRequest(address, value);
-          writer.write(writeReq).get();
-        }
-        catch (Exception e) {
-          logger.error("writing to plc device {} {} failed", connection, address, e);
-        }
-      }
-      
-    };
-  }
+        };
+    }
 
-  static void checkDatatype(Class<?> cls) {
-    if (cls == Boolean.class
-        || cls == Byte.class
-        || cls == Short.class
-        || cls == Integer.class
-        || cls == Float.class
-        || cls == String.class
-        || cls == Calendar.class)
-      return;
-    throw new IllegalArgumentException("Not a legal plc data type: "+cls.getSimpleName());
-  }
+    <T> Consumer<T> newConsumer(Class<T> datatype, String addressStr) {
+        PlcConnectionAdapter.checkDatatype(datatype);
+        return new Consumer<T>() {
+            private static final long serialVersionUID = 1L;
 
-  @SuppressWarnings("unchecked")
-  static <T> PlcWriteRequest newPlcWriteRequest(Address address, T value) {
-    Class<?> cls = value.getClass();
-    return new PlcWriteRequest(cls, address, value);
-  }
+            @Override
+            public void accept(T arg0) {
+                PlcConnection connection = null;
+                Address address = null;
+                try {
+                    connection = getConnection();
+                    address = connection.parseAddress(addressStr);
+                    PlcWriter writer = connection.getWriter().get();
+                    PlcWriteRequest writeReq = PlcConnectionAdapter.newPlcWriteRequest(address, arg0);
+                    writer.write(writeReq).get();
+                } catch (Exception e) {
+                    logger.error("writing to plc device {} {} failed", connection, address, e);
+                }
+            }
 
-  @SuppressWarnings("unchecked")
-  static <T> PlcReadRequest newPlcReadRequest(Class<T> datatype, Address address) {
-      return new PlcReadRequest(datatype, address);
-  }
+        };
+    }
+
+    <T> Consumer<JsonObject> newConsumer(Class<T> datatype, Function<JsonObject, String> addressFn, Function<JsonObject, T> valueFn) {
+        PlcConnectionAdapter.checkDatatype(datatype);
+        return new Consumer<JsonObject>() {
+            private static final long serialVersionUID = 1L;
+
+            @Override
+            public void accept(JsonObject jo) {
+                PlcConnection connection = null;
+                Address address = null;
+                try {
+                    connection = getConnection();
+                    String addressStr = addressFn.apply(jo);
+                    address = connection.parseAddress(addressStr);
+                    T value = valueFn.apply(jo);
+                    PlcWriter writer = connection.getWriter().get();
+                    PlcWriteRequest writeReq = newPlcWriteRequest(address, value);
+                    writer.write(writeReq).get();
+                } catch (Exception e) {
+                    logger.error("writing to plc device {} {} failed", connection, address, e);
+                }
+            }
+
+        };
+    }
+
+    static void checkDatatype(Class<?> cls) {
+        if (cls == Boolean.class
+            || cls == Byte.class
+            || cls == Short.class
+            || cls == Integer.class
+            || cls == Float.class
+            || cls == String.class
+            || cls == Calendar.class)
+            return;
+        throw new IllegalArgumentException("Not a legal plc data type: " + cls.getSimpleName());
+    }
+
+    @SuppressWarnings("unchecked")
+    static <T> PlcWriteRequest<T> newPlcWriteRequest(Address address, T value) {
+        Class<T> cls = (Class<T>) value.getClass();
+        return new PlcWriteRequest<>(cls, address, value);
+    }
+
+    @SuppressWarnings("unchecked")
+    static <T> PlcReadRequest<T> newPlcReadRequest(Class<T> datatype, Address address) {
+        return new PlcReadRequest<>(datatype, address);
+    }
 
 }
diff --git a/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/PlcConnectionAdapterTest.java b/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/PlcConnectionAdapterTest.java
index 88cec7d..36e4d1c 100644
--- a/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/PlcConnectionAdapterTest.java
+++ b/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/PlcConnectionAdapterTest.java
@@ -18,11 +18,7 @@
 */
 package org.apache.plc4x.edgent;
 
-import java.lang.reflect.Array;
-import java.util.Calendar;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-
+import com.google.gson.JsonObject;
 import org.apache.edgent.function.Consumer;
 import org.apache.edgent.function.Function;
 import org.apache.edgent.function.Supplier;
@@ -30,7 +26,10 @@
 import org.apache.plc4x.edgent.mock.MockConnection;
 import org.apache.plc4x.java.PlcDriverManager;
 import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
-import org.apache.plc4x.java.api.messages.*;
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.api.messages.PlcWriteRequest;
+import org.apache.plc4x.java.api.messages.PlcWriteResponse;
 import org.apache.plc4x.java.api.messages.items.ReadRequestItem;
 import org.apache.plc4x.java.api.messages.items.WriteRequestItem;
 import org.apache.plc4x.java.api.model.Address;
@@ -38,507 +37,525 @@
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 
-import com.google.gson.JsonObject;
+import java.lang.reflect.Array;
+import java.util.Calendar;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 
 public class PlcConnectionAdapterTest {
-  
-  // TODO figure out how to get these run via Eclipse (org.junit.jupiter.api?) and remove this
-  // Ah... Junit 5... needs newer Eclipse (newer than neon 1.a)
-  public static void main(String[] args) throws Exception {
-    PlcConnectionAdapterTest t = new PlcConnectionAdapterTest();
-    t.testCtor1();
-    t.testCtor2();
-    t.testCheckDatatype();
-    t.testNewPlcReadRequest();
-    t.testNewPlcWriteRequest();
-    t.testNewSupplier();
-    t.testNewSupplierNeg();
-    t.testNewConsumer1();
-    t.testNewConsumer1Neg();
-    t.testNewConsumer2();
-    t.testNewConsumer2Neg();
-    System.out.println("SUCCESS");
-  }
-  
-  protected MockConnection getMockConnection() throws PlcConnectionException {
-    return (MockConnection) new PlcDriverManager().getConnection("mock://some-cool-url");
-  }
-  
-  /*
-   * Test the PlcConnectionAdapter(PlcConnection) ctor, getConnection() and close()
-   */
-  @Test
-  @Tag("fast")
-  public void testCtor1() throws Exception {
-    MockConnection mockConnection = getMockConnection();
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(mockConnection);
-    Assertions.assertSame(mockConnection, adapter.getConnection());
-    // and again... multiple adapter.getConnection() returns the same
-    Assertions.assertSame(mockConnection, adapter.getConnection());
-    adapter.close();
-  }
-  
-  /*
-   * Test the PlcConnectionAdapter(url) ctor, getConnection() and close()
-   */
-  @Test
-  @Tag("fast")
-  public void testCtor2() throws Exception {
-    MockConnection mockConnection = getMockConnection();
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(mockConnection.getUrl());
-    MockConnection mockConnection2 = (MockConnection) adapter.getConnection();
-    Assertions.assertNotSame(mockConnection, mockConnection2);
-    Assertions.assertSame(mockConnection.getUrl(), mockConnection2.getUrl());
-    // and again... multiple adapter.getConnection() returns the same
-    Assertions.assertSame(mockConnection2, adapter.getConnection());
-    adapter.close();
-  }
- 
-  @Test
-  @Tag("fast")
-  public void testCheckDatatype() throws Exception {
-    PlcConnectionAdapter.checkDatatype(Boolean.class);
-    PlcConnectionAdapter.checkDatatype(Byte.class);
-    PlcConnectionAdapter.checkDatatype(Short.class);
-    PlcConnectionAdapter.checkDatatype(Integer.class);
-    PlcConnectionAdapter.checkDatatype(Float.class);
-    PlcConnectionAdapter.checkDatatype(String.class);
-    PlcConnectionAdapter.checkDatatype(Calendar.class);
-    Assertions.assertThrows(IllegalArgumentException.class,
-        () -> PlcConnectionAdapter.checkDatatype(Long.class));
-    Assertions.assertThrows(IllegalArgumentException.class,
-        () -> PlcConnectionAdapter.checkDatatype(Double.class));
-  }
-  
-  private <T> void checkRead(MockConnection connection, PlcReadRequest request, T value) throws InterruptedException, ExecutionException {
-    // this is really a tests of our mock tooling but knowing it's behaving as expected
-    // will help identify problems in the adapter/supplier/consumer
-    connection.setDataValue(request.getReadRequestItems().get(0).getAddress(), value);
-    
-    CompletableFuture<PlcReadResponse> cf = connection.read(request);
-    
-    Assertions.assertTrue(cf.isDone());
-    PlcReadResponse response = cf.get();
-    Assertions.assertEquals(value, response.getResponseItems().get(0).getValues().get(0));
-  }
-  
-  private <T> void checkWrite(MockConnection connection, PlcWriteRequest request, T value) throws InterruptedException, ExecutionException {
-    // this is really a tests of our mock tooling but knowing it's behaving as expected
-    // will help identify problems in the adapter/supplier/consumer
-    connection.setDataValue(request.getRequestItems().get(0).getAddress(), value);
-    
-    CompletableFuture<PlcWriteResponse> cf = connection.write(request);
-    
-    Assertions.assertTrue(cf.isDone());
-    PlcWriteResponse response = cf.get();
-    Assertions.assertNotNull(response);
-    T writtenData = (T) connection.getDataValue(request.getRequestItems().get(0).getAddress());
-    if(writtenData.getClass().isArray()) {
-      writtenData = (T) Array.get(writtenData, 0);
+
+    // TODO figure out how to get these run via Eclipse (org.junit.jupiter.api?) and remove this
+    // Ah... Junit 5... needs newer Eclipse (newer than neon 1.a)
+    public static void main(String[] args) throws Exception {
+        PlcConnectionAdapterTest t = new PlcConnectionAdapterTest();
+        t.testCtor1();
+        t.testCtor2();
+        t.testCheckDatatype();
+        t.testNewPlcReadRequest();
+        t.testNewPlcWriteRequest();
+        t.testNewSupplier();
+        t.testNewSupplierNeg();
+        t.testNewConsumer1();
+        t.testNewConsumer1Neg();
+        t.testNewConsumer2();
+        t.testNewConsumer2Neg();
+        System.out.println("SUCCESS");
     }
-    Assertions.assertEquals(value, writtenData);
-  }
-  
-  /*
-   * Verify the adapter yields the appropriate PlcReadRequest for each type and that it works.
-   */
-  @SuppressWarnings("unchecked")
-  @Test
-  @Tag("fast")
-  public void testNewPlcReadRequest() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
 
-    PlcReadRequest request;
-    
-    request = PlcConnectionAdapter.newPlcReadRequest(Boolean.class, address);
-    ReadRequestItem requestItem = request.getReadRequestItems().get(0);
-    Class dataType = requestItem.getDatatype();
-    Assertions.assertTrue(dataType == Boolean.class, "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkRead(connection, request, true);
-    checkRead(connection, request, false);
-    
-    request = PlcConnectionAdapter.newPlcReadRequest(Byte.class, address);
-    requestItem = request.getReadRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(dataType == Byte.class, "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkRead(connection, request, (byte)0x13);
-    checkRead(connection, request, (byte)0x23);
-    
-    request = PlcConnectionAdapter.newPlcReadRequest(Short.class, address);
-    requestItem = request.getReadRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(dataType == Short.class, "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkRead(connection, request, (short)13);
-    checkRead(connection, request, (short)23);
-
-    request = PlcConnectionAdapter.newPlcReadRequest(Integer.class, address);
-    requestItem = request.getReadRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(dataType == Integer.class, "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkRead(connection, request, 33);
-    checkRead(connection, request, -133);
-    
-    request = PlcConnectionAdapter.newPlcReadRequest(Float.class, address);
-    requestItem = request.getReadRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(dataType == Float.class, "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkRead(connection, request, 43.5f);
-    checkRead(connection, request, -143.5f);
-    
-    request = PlcConnectionAdapter.newPlcReadRequest(String.class, address);
-    requestItem = request.getReadRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(dataType == String.class, "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkRead(connection, request, "ReadySetGo");
-    checkRead(connection, request, "OneMoreTime");
-    
-    request = PlcConnectionAdapter.newPlcReadRequest(Calendar.class, address);
-    requestItem = request.getReadRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(dataType == Calendar.class, "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkRead(connection, request, Calendar.getInstance());
-    
-    adapter.close();
-  }
-  
-  
-  /*
-   * Verify the adapter yields the appropriate PlcWriteRequest for each type and that it works.
-   */
-  @SuppressWarnings("unchecked")
-  @Test
-  @Tag("fast")
-  public void testNewPlcWriteRequest() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
-
-    PlcWriteRequest request;
-    
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, true);
-    WriteRequestItem requestItem = request.getRequestItems().get(0);
-    Class dataType = requestItem.getDatatype();
-    Assertions.assertTrue(Boolean.class.isAssignableFrom(dataType), "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkWrite(connection, request, true);
-    
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, (byte)0x113);
-    requestItem = request.getRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(Byte.class.isAssignableFrom(dataType), "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkWrite(connection, request, (byte)0x113);
-    
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, (short)113);
-    requestItem = request.getRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(Short.class.isAssignableFrom(dataType), "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkWrite(connection, request, (short)113);
-
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, 1033);
-    requestItem = request.getRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(Integer.class.isAssignableFrom(dataType), "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkWrite(connection, request, 1033);
-    
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, 1043.5f);
-    requestItem = request.getRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(Float.class.isAssignableFrom(dataType), "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkWrite(connection, request, 1043.5f);
-    
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, "A written value");
-    requestItem = request.getRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    Assertions.assertTrue(String.class.isAssignableFrom(dataType), "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkWrite(connection, request, "A written value");
-    
-    Calendar calValue = Calendar.getInstance();
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, calValue);
-    requestItem = request.getRequestItems().get(0);
-    dataType = requestItem.getDatatype();
-    request = PlcConnectionAdapter.newPlcWriteRequest(address, calValue);
-    Assertions.assertTrue(Calendar.class.isAssignableFrom(dataType), "class:"+request.getClass());
-    Assertions.assertSame(address, requestItem.getAddress());
-    checkWrite(connection, request, calValue);
-    
-    adapter.close();
-  }
-
-  /*
-   * test PlcConnectionAdapter.newSupplier
-   */
-  @SuppressWarnings("unchecked")
-  @Test
-  @Tag("fast")
-  public void testNewSupplier() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
-
-    Supplier<?> supplier;
-    
-    supplier = adapter.newSupplier(Boolean.class, addressStr);
-    Assertions.assertNotSame(supplier, adapter.newSupplier(Boolean.class, addressStr));
-    checkSupplier(connection, address, (Supplier<Boolean>)supplier, true, false);
-    
-    supplier = adapter.newSupplier(Byte.class, addressStr);
-    checkSupplier(connection, address, (Supplier<Byte>)supplier, (byte)0x1, (byte)0x2, (byte)0x3);
-    
-    supplier = adapter.newSupplier(Short.class, addressStr);
-    checkSupplier(connection, address, (Supplier<Short>)supplier, (short)1, (short)2, (short)3);
-
-    supplier = adapter.newSupplier(Integer.class, addressStr);
-    checkSupplier(connection, address, (Supplier<Integer>)supplier, 1000, 1001, 1002);
-    
-    supplier = adapter.newSupplier(Float.class, addressStr);
-    checkSupplier(connection, address, (Supplier<Float>)supplier, 1000.5f, 1001.5f, 1002.5f);
-    
-    supplier = adapter.newSupplier(String.class, addressStr);
-    checkSupplier(connection, address, (Supplier<String>)supplier, "one", "two", "three");
-    
-    adapter.close();
-  }
-  
-  /*
-   * test PlcConnectionAdapter.newSupplier with read exception
-   */
-  @SuppressWarnings("unchecked")
-  @Test
-  @Tag("fast")
-  public void testNewSupplierNeg() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
-
-    Supplier<?> supplier;
-    
-    supplier = adapter.newSupplier(String.class, addressStr);
-    checkSupplier(2, connection, address, (Supplier<String>)supplier, "one", "two", "three");
-    
-    adapter.close();
-  }
-  
-  static <T> void checkSupplier(MockConnection connection, Address address, Supplier<T> supplier, Object ... values) throws Exception {
-    checkSupplier(0, connection, address, supplier, values);
-  }
-  private static <T> void checkSupplier(int readFailureCountTrigger, MockConnection connection, Address address, Supplier<T> supplier, Object ... values) throws Exception {
-    // verify that a read failure doesn't kill the consumer
-    // it logs (not verified) but returns null (as designed) and keeps working for the subsequent reads
-    connection.setReadException(readFailureCountTrigger, "This is a mock read exception");
-    int readCount = 0;
-    for (Object value : values) {
-      connection.setDataValue(address, value);
-      T readData = supplier.get();
-      // System.out.println("checkSupplier"+(readFailureCountTrigger > 0 ? "NEG" : "")+": value:"+value+" readData:"+readData);
-      if (readFailureCountTrigger <= 0)
-        Assertions.assertEquals(value, readData);
-      else {
-        if (++readCount != readFailureCountTrigger)
-          Assertions.assertEquals(value, readData);
-        else
-          Assertions.assertNull(readData);
-      }
+    protected MockConnection getMockConnection() throws PlcConnectionException {
+        return (MockConnection) new PlcDriverManager().getConnection("mock://some-cool-url");
     }
-  }
 
-  /*
-   * test PlcConnectionAdapter.newConsumer(address)
-   */
-  @SuppressWarnings("unchecked")
-  @Test
-  @Tag("fast")
-  public void testNewConsumer1() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
+    /*
+     * Test the PlcConnectionAdapter(PlcConnection) ctor, getConnection() and close()
+     */
+    @Test
+    @Tag("fast")
+    public void testCtor1() throws Exception {
+        MockConnection mockConnection = getMockConnection();
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(mockConnection);
+        Assertions.assertSame(mockConnection, adapter.getConnection());
+        // and again... multiple adapter.getConnection() returns the same
+        Assertions.assertSame(mockConnection, adapter.getConnection());
+        adapter.close();
+    }
 
-    Consumer<?> consumer;
-    
-    consumer = adapter.newConsumer(Boolean.class, addressStr);
-    Assertions.assertNotSame(consumer, adapter.newConsumer(Boolean.class, addressStr));
-    checkConsumer(connection, address, consumer, true, false);
-    
-    consumer = adapter.newConsumer(Byte.class, addressStr);
-    checkConsumer(connection, address, consumer, (byte)0x1, (byte)0x2, (byte)0x3);
-    
-    consumer = adapter.newConsumer(Short.class, addressStr);
-    checkConsumer(connection, address, consumer, (short)1, (short)2, (short)3);
+    /*
+     * Test the PlcConnectionAdapter(url) ctor, getConnection() and close()
+     */
+    @Test
+    @Tag("fast")
+    public void testCtor2() throws Exception {
+        MockConnection mockConnection = getMockConnection();
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(mockConnection.getUrl());
+        MockConnection mockConnection2 = (MockConnection) adapter.getConnection();
+        Assertions.assertNotSame(mockConnection, mockConnection2);
+        Assertions.assertSame(mockConnection.getUrl(), mockConnection2.getUrl());
+        // and again... multiple adapter.getConnection() returns the same
+        Assertions.assertSame(mockConnection2, adapter.getConnection());
+        adapter.close();
+    }
 
-    consumer = adapter.newConsumer(Integer.class, addressStr);
-    checkConsumer(connection, address, consumer, 1000, 1001, 1002);
-    
-    consumer = adapter.newConsumer(Float.class, addressStr);
-    checkConsumer(connection, address, consumer, 1000.5f, 1001.5f, 1002.5f);
-    
-    consumer = adapter.newConsumer(String.class, addressStr);
-    checkConsumer(connection, address, consumer, "one", "two", "three");
-    
-    adapter.close();
-  }
+    @Test
+    @Tag("fast")
+    public void testCheckDatatype() throws Exception {
+        PlcConnectionAdapter.checkDatatype(Boolean.class);
+        PlcConnectionAdapter.checkDatatype(Byte.class);
+        PlcConnectionAdapter.checkDatatype(Short.class);
+        PlcConnectionAdapter.checkDatatype(Integer.class);
+        PlcConnectionAdapter.checkDatatype(Float.class);
+        PlcConnectionAdapter.checkDatatype(String.class);
+        PlcConnectionAdapter.checkDatatype(Calendar.class);
+        Assertions.assertThrows(IllegalArgumentException.class,
+            () -> PlcConnectionAdapter.checkDatatype(Long.class));
+        Assertions.assertThrows(IllegalArgumentException.class,
+            () -> PlcConnectionAdapter.checkDatatype(Double.class));
+    }
 
-  /*
-   * test PlcConnectionAdapter.newConsumer(address) with write exception
-   */
-  @SuppressWarnings("unchecked")
-  @Test
-  @Tag("fast")
-  public void testNewConsumer1Neg() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
+    private <T> void checkRead(MockConnection connection, PlcReadRequest<T> request, T value) throws InterruptedException, ExecutionException {
+        // this is really a tests of our mock tooling but knowing it's behaving as expected
+        // will help identify problems in the adapter/supplier/consumer
+        connection.setDataValue(request.getReadRequestItems().get(0).getAddress(), value);
 
-    Consumer<?> consumer;
-    
-    consumer = adapter.newConsumer(String.class, addressStr);
-    checkConsumer(2, connection, address, (Consumer<String>)consumer, "one", "two", "three");
-    
-    adapter.close();
-  }
-  
-  static <T> void checkConsumer(MockConnection connection, Address address, Consumer<T> consumer, Object ... values) throws Exception {
-    checkConsumer(0, connection, address, consumer, values);
-  }
-  @SuppressWarnings("unchecked")
-  private static <T> void checkConsumer(int writeFailureCountTrigger, MockConnection connection, Address address, Consumer<T> consumer, Object ... values) throws Exception {
-    // verify that a write failure doesn't kill the consumer
-    // it logs (not verified) but keeps working for the subsequent writes
-    connection.setWriteException(writeFailureCountTrigger, "This is a mock write exception");
-    int writeCount = 0;
-    Object previousValue = null;
-    for (Object value : values) {
-      consumer.accept((T)value);
-      T writtenData = (T) connection.getDataValue(address);
-      if(writtenData.getClass().isArray()) {
-        writtenData = (T) Array.get(writtenData, 0);
-      }
-      // System.out.println("checkConsumer"+(writeFailureCountTrigger > 0 ? "NEG" : "")+": value:"+value+" writtenData:"+writtenData);
-      if (writeFailureCountTrigger <= 0)
+        CompletableFuture<PlcReadResponse<T>> cf = connection.read(request);
+
+        Assertions.assertTrue(cf.isDone());
+        PlcReadResponse<T> response = cf.get();
+        Assertions.assertEquals(value, response.getResponseItems().get(0).getValues().get(0));
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T> void checkWrite(MockConnection connection, PlcWriteRequest<T> request, T value) throws InterruptedException, ExecutionException {
+        // this is really a tests of our mock tooling but knowing it's behaving as expected
+        // will help identify problems in the adapter/supplier/consumer
+        connection.setDataValue(request.getRequestItems().get(0).getAddress(), value);
+
+        CompletableFuture<PlcWriteResponse<T>> cf = connection.write(request);
+
+        Assertions.assertTrue(cf.isDone());
+        PlcWriteResponse response = cf.get();
+        Assertions.assertNotNull(response);
+        T writtenData = (T) connection.getDataValue(request.getRequestItems().get(0).getAddress());
+        if (writtenData.getClass().isArray()) {
+            writtenData = (T) Array.get(writtenData, 0);
+        }
         Assertions.assertEquals(value, writtenData);
-      else { 
-        if (++writeCount != writeFailureCountTrigger)
-          Assertions.assertEquals(value, writtenData);
-        else
-          Assertions.assertEquals(previousValue, writtenData);
-      }
-      previousValue = value;
     }
-  }
 
-  /*
-   * test PlcConnectionAdapter.newConsumer(addressFn, valueFn)
-   */
-  @Test
-  @Tag("fast")
-  public void testNewConsumer2() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
+    /*
+     * Verify the adapter yields the appropriate PlcReadRequest for each type and that it works.
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    @Tag("fast")
+    public void testNewPlcReadRequest() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
 
-    Consumer<JsonObject> consumer;
-    
-    Function<JsonObject,String> addressFn = t -> t.get("address").getAsString(); 
-    
-    consumer = adapter.newConsumer(Boolean.class, addressFn, t -> t.get("value").getAsBoolean());
-    checkConsumerJson(connection, address, consumer, true, false);
-    
-    consumer = adapter.newConsumer(Byte.class, addressFn, t -> t.get("value").getAsByte());
-    checkConsumerJson(connection, address, consumer, (byte)0x1, (byte)0x2, (byte)0x3);
-    
-    consumer = adapter.newConsumer(Short.class, addressFn, t -> t.get("value").getAsShort());
-    checkConsumerJson(connection, address, consumer, (short)1, (short)2, (short)3);
-
-    consumer = adapter.newConsumer(Integer.class, addressFn, t -> t.get("value").getAsInt());
-    checkConsumerJson(connection, address, consumer, 1000, 1001, 1002);
-    
-    consumer = adapter.newConsumer(Float.class, addressFn, t -> t.get("value").getAsFloat());
-    checkConsumerJson(connection, address, consumer, 1000.5f, 1001.5f, 1002.5f);
-    
-    consumer = adapter.newConsumer(String.class, addressFn, t -> t.get("value").getAsString());
-    checkConsumerJson(connection, address, consumer, "one", "two", "three");
-    
-    adapter.close();
-  }
-
-  /*
-   * test PlcConnectionAdapter.newConsumer(addressFn, valueFn) with write failure
-   */
-  @Test
-  @Tag("fast")
-  public void testNewConsumer2Neg() throws Exception {
-    String addressStr = "MyReadWriteAddress/0";
-    MockAddress address = new MockAddress(addressStr);
-    PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
-    MockConnection connection = (MockConnection) adapter.getConnection();
-
-    Consumer<JsonObject> consumer;
-    
-    Function<JsonObject,String> addressFn = t -> t.get("address").getAsString(); 
-    
-    consumer = adapter.newConsumer(String.class, addressFn, t -> t.get("value").getAsString());
-    checkConsumerJson(2, connection, address, consumer, "one", "two", "three");
-    
-    adapter.close();
-  }
-  
-  static <T> void checkConsumerJson(MockConnection connection, MockAddress address, Consumer<JsonObject> consumer, Object ... values) throws Exception {
-    checkConsumerJson(0, connection, address, consumer, values);
-  }
-  private static <T> void checkConsumerJson(int writeFailureCountTrigger, MockConnection connection, MockAddress address, Consumer<JsonObject> consumer, Object ... values) throws Exception {
-    if (writeFailureCountTrigger > 0)
-      connection.setWriteException(writeFailureCountTrigger, "This is a mock write exception");
-    int writeCount = 0;
-    Object previousValue = null;
-    for (Object value : values) {
-      
-      // build the JsonObject to consume
-      JsonObject jo = new JsonObject();
-      jo.addProperty("address", address.getAddress());
-      if (value instanceof Boolean)
-        jo.addProperty("value", (Boolean)value);
-      else if (value instanceof Number)
-        jo.addProperty("value", (Number)value);
-      else if (value instanceof String)
-        jo.addProperty("value", (String)value);
-      
-      consumer.accept(jo);
-      
-      @SuppressWarnings("unchecked")
-      T writtenData = (T) connection.getDataValue(address);
-      if(writtenData.getClass().isArray()) {
-        writtenData = (T) Array.get(writtenData, 0);
-      }
-      // System.out.println("checkConsumerJson"+(writeFailureCountTrigger > 0 ? "NEG" : "")+": value:"+value+" writtenData:"+writtenData);
-      if (writeFailureCountTrigger <= 0)
-        Assertions.assertEquals(value, writtenData);
-      else { 
-        if (++writeCount != writeFailureCountTrigger)
-          Assertions.assertEquals(value, writtenData);
-        else
-          Assertions.assertEquals(previousValue, writtenData);
-      }
-      previousValue = value;
+        {
+            PlcReadRequest<Boolean> request = PlcConnectionAdapter.newPlcReadRequest(Boolean.class, address);
+            ReadRequestItem<Boolean> requestItem = request.getReadRequestItems().get(0);
+            Class<Boolean> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(dataType == Boolean.class, "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkRead(connection, request, true);
+            checkRead(connection, request, false);
+        }
+        {
+            PlcReadRequest<Byte> request = PlcConnectionAdapter.newPlcReadRequest(Byte.class, address);
+            ReadRequestItem<Byte> requestItem = request.getReadRequestItems().get(0);
+            Class<Byte> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(dataType == Byte.class, "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkRead(connection, request, (byte) 0x13);
+            checkRead(connection, request, (byte) 0x23);
+        }
+        {
+            PlcReadRequest<Short> request = PlcConnectionAdapter.newPlcReadRequest(Short.class, address);
+            ReadRequestItem<Short> requestItem = request.getReadRequestItems().get(0);
+            Class<Short> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(dataType == Short.class, "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkRead(connection, request, (short) 13);
+            checkRead(connection, request, (short) 23);
+        }
+        {
+            PlcReadRequest<Integer> request = PlcConnectionAdapter.newPlcReadRequest(Integer.class, address);
+            ReadRequestItem<Integer> requestItem = request.getReadRequestItems().get(0);
+            Class<Integer> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(dataType == Integer.class, "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkRead(connection, request, 33);
+            checkRead(connection, request, -133);
+        }
+        {
+            PlcReadRequest<Float> request = PlcConnectionAdapter.newPlcReadRequest(Float.class, address);
+            ReadRequestItem<Float> requestItem = request.getReadRequestItems().get(0);
+            Class<Float> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(dataType == Float.class, "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkRead(connection, request, 43.5f);
+            checkRead(connection, request, -143.5f);
+        }
+        {
+            PlcReadRequest<String> request = PlcConnectionAdapter.newPlcReadRequest(String.class, address);
+            ReadRequestItem<String> requestItem = request.getReadRequestItems().get(0);
+            Class<String> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(dataType == String.class, "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkRead(connection, request, "ReadySetGo");
+            checkRead(connection, request, "OneMoreTime");
+        }
+        {
+            PlcReadRequest<Calendar> request = PlcConnectionAdapter.newPlcReadRequest(Calendar.class, address);
+            ReadRequestItem<Calendar> requestItem = request.getReadRequestItems().get(0);
+            Class<Calendar> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(dataType == Calendar.class, "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkRead(connection, request, Calendar.getInstance());
+        }
+        adapter.close();
     }
-  }
+
+
+    /*
+     * Verify the adapter yields the appropriate PlcWriteRequest for each type and that it works.
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    @Tag("fast")
+    public void testNewPlcWriteRequest() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
+
+        {
+            PlcWriteRequest<Boolean> request = PlcConnectionAdapter.newPlcWriteRequest(address, true);
+            WriteRequestItem<Boolean> requestItem = request.getRequestItems().get(0);
+            Class<Boolean> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(Boolean.class.isAssignableFrom(dataType), "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkWrite(connection, request, true);
+        }
+        {
+            PlcWriteRequest<Byte> request = PlcConnectionAdapter.newPlcWriteRequest(address, (byte) 0x113);
+            WriteRequestItem<Byte> requestItem = request.getRequestItems().get(0);
+            Class<Byte> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(Byte.class.isAssignableFrom(dataType), "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkWrite(connection, request, (byte) 0x113);
+        }
+        {
+            PlcWriteRequest<Short> request = PlcConnectionAdapter.newPlcWriteRequest(address, (short) 113);
+            WriteRequestItem<Short> requestItem = request.getRequestItems().get(0);
+            Class<Short> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(Short.class.isAssignableFrom(dataType), "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkWrite(connection, request, (short) 113);
+        }
+        {
+            PlcWriteRequest<Integer> request = PlcConnectionAdapter.newPlcWriteRequest(address, 1033);
+            WriteRequestItem<Integer> requestItem = request.getRequestItems().get(0);
+            Class<Integer> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(Integer.class.isAssignableFrom(dataType), "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkWrite(connection, request, 1033);
+        }
+        {
+            PlcWriteRequest<Float> request = PlcConnectionAdapter.newPlcWriteRequest(address, 1043.5f);
+            WriteRequestItem<Float> requestItem = request.getRequestItems().get(0);
+            Class<Float> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(Float.class.isAssignableFrom(dataType), "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkWrite(connection, request, 1043.5f);
+        }
+        {
+            PlcWriteRequest<String> request = PlcConnectionAdapter.newPlcWriteRequest(address, "A written value");
+            WriteRequestItem<String> requestItem = request.getRequestItems().get(0);
+            Class<String> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(String.class.isAssignableFrom(dataType), "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkWrite(connection, request, "A written value");
+        }
+        {
+            Calendar calValue = Calendar.getInstance();
+            PlcWriteRequest<Calendar> request = PlcConnectionAdapter.newPlcWriteRequest(address, calValue);
+            WriteRequestItem<Calendar> requestItem = request.getRequestItems().get(0);
+            Class<Calendar> dataType = requestItem.getDatatype();
+            Assertions.assertTrue(Calendar.class.isAssignableFrom(dataType), "class:" + request.getClass());
+            Assertions.assertSame(address, requestItem.getAddress());
+            checkWrite(connection, request, calValue);
+        }
+        adapter.close();
+    }
+
+    /*
+     * test PlcConnectionAdapter.newSupplier
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    @Tag("fast")
+    public void testNewSupplier() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
+
+        {
+            Supplier<Boolean> supplier = adapter.newSupplier(Boolean.class, addressStr);
+            Assertions.assertNotSame(supplier, adapter.newSupplier(Boolean.class, addressStr));
+            checkSupplier(connection, address, supplier, true, false);
+        }
+        {
+            Supplier<Byte> supplier = adapter.newSupplier(Byte.class, addressStr);
+            checkSupplier(connection, address, supplier, (byte) 0x1, (byte) 0x2, (byte) 0x3);
+        }
+        {
+            Supplier<Short> supplier = adapter.newSupplier(Short.class, addressStr);
+            checkSupplier(connection, address, supplier, (short) 1, (short) 2, (short) 3);
+        }
+        {
+            Supplier<Integer> supplier = adapter.newSupplier(Integer.class, addressStr);
+            checkSupplier(connection, address, supplier, 1000, 1001, 1002);
+        }
+        {
+            Supplier<Float> supplier = adapter.newSupplier(Float.class, addressStr);
+            checkSupplier(connection, address, supplier, 1000.5f, 1001.5f, 1002.5f);
+        }
+        {
+            Supplier<?> supplier = adapter.newSupplier(String.class, addressStr);
+            checkSupplier(connection, address, supplier, "one", "two", "three");
+        }
+        adapter.close();
+    }
+
+    /*
+     * test PlcConnectionAdapter.newSupplier with read exception
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    @Tag("fast")
+    public void testNewSupplierNeg() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
+
+        Supplier<String> supplier = adapter.newSupplier(String.class, addressStr);
+        checkSupplier(2, connection, address, supplier, "one", "two", "three");
+
+        adapter.close();
+    }
+
+    static <T> void checkSupplier(MockConnection connection, Address address, Supplier<T> supplier, Object... values) throws Exception {
+        checkSupplier(0, connection, address, supplier, values);
+    }
+
+    private static <T> void checkSupplier(int readFailureCountTrigger, MockConnection connection, Address address, Supplier<T> supplier, Object... values) throws Exception {
+        // verify that a read failure doesn't kill the consumer
+        // it logs (not verified) but returns null (as designed) and keeps working for the subsequent reads
+        connection.setReadException(readFailureCountTrigger, "This is a mock read exception");
+        int readCount = 0;
+        for (Object value : values) {
+            connection.setDataValue(address, value);
+            T readData = supplier.get();
+            // System.out.println("checkSupplier"+(readFailureCountTrigger > 0 ? "NEG" : "")+": value:"+value+" readData:"+readData);
+            if (readFailureCountTrigger <= 0)
+                Assertions.assertEquals(value, readData);
+            else {
+                if (++readCount != readFailureCountTrigger)
+                    Assertions.assertEquals(value, readData);
+                else
+                    Assertions.assertNull(readData);
+            }
+        }
+    }
+
+    /*
+     * test PlcConnectionAdapter.newConsumer(address)
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    @Tag("fast")
+    public void testNewConsumer1() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
+
+        Consumer<?> consumer;
+
+        consumer = adapter.newConsumer(Boolean.class, addressStr);
+        Assertions.assertNotSame(consumer, adapter.newConsumer(Boolean.class, addressStr));
+        checkConsumer(connection, address, consumer, true, false);
+
+        consumer = adapter.newConsumer(Byte.class, addressStr);
+        checkConsumer(connection, address, consumer, (byte) 0x1, (byte) 0x2, (byte) 0x3);
+
+        consumer = adapter.newConsumer(Short.class, addressStr);
+        checkConsumer(connection, address, consumer, (short) 1, (short) 2, (short) 3);
+
+        consumer = adapter.newConsumer(Integer.class, addressStr);
+        checkConsumer(connection, address, consumer, 1000, 1001, 1002);
+
+        consumer = adapter.newConsumer(Float.class, addressStr);
+        checkConsumer(connection, address, consumer, 1000.5f, 1001.5f, 1002.5f);
+
+        consumer = adapter.newConsumer(String.class, addressStr);
+        checkConsumer(connection, address, consumer, "one", "two", "three");
+
+        adapter.close();
+    }
+
+    /*
+     * test PlcConnectionAdapter.newConsumer(address) with write exception
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    @Tag("fast")
+    public void testNewConsumer1Neg() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
+
+        Consumer<?> consumer;
+
+        consumer = adapter.newConsumer(String.class, addressStr);
+        checkConsumer(2, connection, address, (Consumer<String>) consumer, "one", "two", "three");
+
+        adapter.close();
+    }
+
+    static <T> void checkConsumer(MockConnection connection, Address address, Consumer<T> consumer, Object... values) throws Exception {
+        checkConsumer(0, connection, address, consumer, values);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> void checkConsumer(int writeFailureCountTrigger, MockConnection connection, Address address, Consumer<T> consumer, Object... values) throws Exception {
+        // verify that a write failure doesn't kill the consumer
+        // it logs (not verified) but keeps working for the subsequent writes
+        connection.setWriteException(writeFailureCountTrigger, "This is a mock write exception");
+        int writeCount = 0;
+        Object previousValue = null;
+        for (Object value : values) {
+            consumer.accept((T) value);
+            T writtenData = (T) connection.getDataValue(address);
+            if (writtenData.getClass().isArray()) {
+                writtenData = (T) Array.get(writtenData, 0);
+            }
+            // System.out.println("checkConsumer"+(writeFailureCountTrigger > 0 ? "NEG" : "")+": value:"+value+" writtenData:"+writtenData);
+            if (writeFailureCountTrigger <= 0)
+                Assertions.assertEquals(value, writtenData);
+            else {
+                if (++writeCount != writeFailureCountTrigger)
+                    Assertions.assertEquals(value, writtenData);
+                else
+                    Assertions.assertEquals(previousValue, writtenData);
+            }
+            previousValue = value;
+        }
+    }
+
+    /*
+     * test PlcConnectionAdapter.newConsumer(addressFn, valueFn)
+     */
+    @Test
+    @Tag("fast")
+    public void testNewConsumer2() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
+
+        Consumer<JsonObject> consumer;
+
+        Function<JsonObject, String> addressFn = t -> t.get("address").getAsString();
+
+        consumer = adapter.newConsumer(Boolean.class, addressFn, t -> t.get("value").getAsBoolean());
+        checkConsumerJson(connection, address, consumer, true, false);
+
+        consumer = adapter.newConsumer(Byte.class, addressFn, t -> t.get("value").getAsByte());
+        checkConsumerJson(connection, address, consumer, (byte) 0x1, (byte) 0x2, (byte) 0x3);
+
+        consumer = adapter.newConsumer(Short.class, addressFn, t -> t.get("value").getAsShort());
+        checkConsumerJson(connection, address, consumer, (short) 1, (short) 2, (short) 3);
+
+        consumer = adapter.newConsumer(Integer.class, addressFn, t -> t.get("value").getAsInt());
+        checkConsumerJson(connection, address, consumer, 1000, 1001, 1002);
+
+        consumer = adapter.newConsumer(Float.class, addressFn, t -> t.get("value").getAsFloat());
+        checkConsumerJson(connection, address, consumer, 1000.5f, 1001.5f, 1002.5f);
+
+        consumer = adapter.newConsumer(String.class, addressFn, t -> t.get("value").getAsString());
+        checkConsumerJson(connection, address, consumer, "one", "two", "three");
+
+        adapter.close();
+    }
+
+    /*
+     * test PlcConnectionAdapter.newConsumer(addressFn, valueFn) with write failure
+     */
+    @Test
+    @Tag("fast")
+    public void testNewConsumer2Neg() throws Exception {
+        String addressStr = "MyReadWriteAddress/0";
+        MockAddress address = new MockAddress(addressStr);
+        PlcConnectionAdapter adapter = new PlcConnectionAdapter(getMockConnection());
+        MockConnection connection = (MockConnection) adapter.getConnection();
+
+        Consumer<JsonObject> consumer;
+
+        Function<JsonObject, String> addressFn = t -> t.get("address").getAsString();
+
+        consumer = adapter.newConsumer(String.class, addressFn, t -> t.get("value").getAsString());
+        checkConsumerJson(2, connection, address, consumer, "one", "two", "three");
+
+        adapter.close();
+    }
+
+    static <T> void checkConsumerJson(MockConnection connection, MockAddress address, Consumer<JsonObject> consumer, Object... values) throws Exception {
+        checkConsumerJson(0, connection, address, consumer, values);
+    }
+
+    private static <T> void checkConsumerJson(int writeFailureCountTrigger, MockConnection connection, MockAddress address, Consumer<JsonObject> consumer, Object... values) throws Exception {
+        if (writeFailureCountTrigger > 0)
+            connection.setWriteException(writeFailureCountTrigger, "This is a mock write exception");
+        int writeCount = 0;
+        Object previousValue = null;
+        for (Object value : values) {
+
+            // build the JsonObject to consume
+            JsonObject jo = new JsonObject();
+            jo.addProperty("address", address.getAddress());
+            if (value instanceof Boolean)
+                jo.addProperty("value", (Boolean) value);
+            else if (value instanceof Number)
+                jo.addProperty("value", (Number) value);
+            else if (value instanceof String)
+                jo.addProperty("value", (String) value);
+
+            consumer.accept(jo);
+
+            @SuppressWarnings("unchecked")
+            T writtenData = (T) connection.getDataValue(address);
+            if (writtenData.getClass().isArray()) {
+                writtenData = (T) Array.get(writtenData, 0);
+            }
+            // System.out.println("checkConsumerJson"+(writeFailureCountTrigger > 0 ? "NEG" : "")+": value:"+value+" writtenData:"+writtenData);
+            if (writeFailureCountTrigger <= 0)
+                Assertions.assertEquals(value, writtenData);
+            else {
+                if (++writeCount != writeFailureCountTrigger)
+                    Assertions.assertEquals(value, writtenData);
+                else
+                    Assertions.assertEquals(previousValue, writtenData);
+            }
+            previousValue = value;
+        }
+    }
 
 }
diff --git a/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/mock/MockConnection.java b/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/mock/MockConnection.java
index d02c7a0..c81ce11 100644
--- a/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/mock/MockConnection.java
+++ b/integrations/apache-edgent/src/test/java/org/apache/plc4x/edgent/mock/MockConnection.java
@@ -18,9 +18,6 @@
 */
 package org.apache.plc4x.edgent.mock;
 
-import java.util.*;
-import java.util.concurrent.CompletableFuture;
-
 import org.apache.plc4x.java.api.authentication.PlcAuthentication;
 import org.apache.plc4x.java.api.connection.AbstractPlcConnection;
 import org.apache.plc4x.java.api.connection.PlcReader;
@@ -39,33 +36,36 @@
 import org.apache.plc4x.java.api.model.Address;
 import org.apache.plc4x.java.api.types.ResponseCode;
 
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+
 public class MockConnection extends AbstractPlcConnection implements PlcReader, PlcWriter {
 
     private final String url;
     private final PlcAuthentication authentication;
-    private final Map<Address,Object> dataValueMap = new HashMap<>();
+    private final Map<Address, Object> dataValueMap = new HashMap<>();
     private long curReadCnt;
     private int readExceptionTriggerCount;
     private String readExceptionMsg;
     private long curWriteCnt;
     private int writeExceptionTriggerCount;
     private String writeExceptionMsg;
-  
+
     public MockConnection(String url) {
-      this(url, null);
+        this(url, null);
     }
-  
+
     public MockConnection(String url, PlcAuthentication authentication) {
-      this.url = url;
-      this.authentication = authentication;
+        this.url = url;
+        this.authentication = authentication;
     }
-  
+
     public PlcAuthentication getAuthentication() {
         return authentication;
     }
-    
+
     public String getUrl() {
-      return url;
+        return url;
     }
 
     @Override
@@ -80,68 +80,73 @@
 
     @Override
     public Address parseAddress(String addressString) throws PlcException {
-      return new MockAddress(addressString);
+        return new MockAddress(addressString);
     }
 
     @SuppressWarnings("unchecked")
     @Override
-    public CompletableFuture<PlcReadResponse> read(PlcReadRequest readRequest) {
-      curReadCnt++;
-      if (readExceptionTriggerCount > 0 && curReadCnt == readExceptionTriggerCount) {
-        curReadCnt = 0;
-        CompletableFuture<PlcReadResponse> cf = new CompletableFuture<>();
-        cf.completeExceptionally(new PlcIoException(readExceptionMsg));
-        return cf;
-      }
-      List<ReadResponseItem> responseItems = new LinkedList<>();
-      for (ReadRequestItem requestItem : readRequest.getReadRequestItems()) {
-        ReadResponseItem responseItem = new ReadResponseItem(requestItem, ResponseCode.OK,
-          Collections.singletonList(getDataValue(requestItem.getAddress())));
-        responseItems.add(responseItem);
-      }
-      PlcReadResponse response = new PlcReadResponse(readRequest, responseItems);
-      return CompletableFuture.completedFuture(response);
+    public <T> CompletableFuture<PlcReadResponse<T>> read(PlcReadRequest<T> readRequest) {
+        curReadCnt++;
+        if (readExceptionTriggerCount > 0 && curReadCnt == readExceptionTriggerCount) {
+            curReadCnt = 0;
+            CompletableFuture<PlcReadResponse<T>> cf = new CompletableFuture<>();
+            cf.completeExceptionally(new PlcIoException(readExceptionMsg));
+            return cf;
+        }
+        List<ReadResponseItem> responseItems = new LinkedList<>();
+        for (ReadRequestItem requestItem : readRequest.getReadRequestItems()) {
+            ReadResponseItem responseItem = new ReadResponseItem(requestItem, ResponseCode.OK,
+                Collections.singletonList(getDataValue(requestItem.getAddress())));
+            responseItems.add(responseItem);
+        }
+        PlcReadResponse response = new PlcReadResponse(readRequest, responseItems);
+        return CompletableFuture.completedFuture(response);
     }
 
     @Override
-    public CompletableFuture<PlcWriteResponse> write(PlcWriteRequest writeRequest) {
-      curWriteCnt++;
-      if (writeExceptionTriggerCount > 0 && curWriteCnt == writeExceptionTriggerCount) {
-        curWriteCnt = 0;
-        CompletableFuture<PlcWriteResponse> cf = new CompletableFuture<>();
-        cf.completeExceptionally(new PlcIoException(writeExceptionMsg));
-        return cf;
-      }
-       List<WriteResponseItem> responseItems = new LinkedList<>();
-        for (WriteRequestItem requestItem : writeRequest.getRequestItems()) {
-            setDataValue(requestItem.getAddress(), requestItem.getValues());
-            WriteResponseItem responseItem = new WriteResponseItem(requestItem, ResponseCode.OK);
-          responseItems.add(responseItem);
+    public <T> CompletableFuture<PlcWriteResponse<T>> write(PlcWriteRequest<T> writeRequest) {
+        curWriteCnt++;
+        if (writeExceptionTriggerCount > 0 && curWriteCnt == writeExceptionTriggerCount) {
+            curWriteCnt = 0;
+            CompletableFuture<PlcWriteResponse<T>> cf = new CompletableFuture<>();
+            cf.completeExceptionally(new PlcIoException(writeExceptionMsg));
+            return cf;
         }
-      PlcWriteResponse response = new PlcWriteResponse(writeRequest, responseItems);
-      return CompletableFuture.completedFuture(response);
+        List<WriteResponseItem<T>> responseItems = new LinkedList<>();
+        for (WriteRequestItem<T> requestItem : writeRequest.getRequestItems()) {
+            setDataValue(requestItem.getAddress(), requestItem.getValues());
+            WriteResponseItem<T> responseItem = new WriteResponseItem<>(requestItem, ResponseCode.OK);
+            responseItems.add(responseItem);
+        }
+        PlcWriteResponse<T> response = new PlcWriteResponse<>(writeRequest, responseItems);
+        return CompletableFuture.completedFuture(response);
     }
 
     public void setDataValue(Address address, Object o) {
-      dataValueMap.put(address, o);
+        dataValueMap.put(address, o);
     }
+
     public Object getDataValue(Address address) {
-      return dataValueMap.get(address);
+        return dataValueMap.get(address);
     }
-    public Map<Address,Object> getAllDataValues() {
-      return dataValueMap;
+
+    public Map<Address, Object> getAllDataValues() {
+        return dataValueMap;
     }
+
     public void clearAllDataValues() {
-      dataValueMap.clear();
+        dataValueMap.clear();
     }
+
     public void setReadException(int readTriggerCount, String msg) {
-      readExceptionTriggerCount = readTriggerCount;
-      readExceptionMsg = msg;
-      curReadCnt = 0;
+        readExceptionTriggerCount = readTriggerCount;
+        readExceptionMsg = msg;
+        curReadCnt = 0;
     }
+
     public void setWriteException(int writeTriggerCount, String msg) {
-      writeExceptionTriggerCount = writeTriggerCount;
-      writeExceptionMsg = msg;
-      curWriteCnt = 0;
+        writeExceptionTriggerCount = writeTriggerCount;
+        writeExceptionMsg = msg;
+        curWriteCnt = 0;
     }
 }
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcReader.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcReader.java
index 35ddf1b..d6faa78 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcReader.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcReader.java
@@ -28,12 +28,14 @@
  */
 public interface PlcReader {
 
+
     /**
      * Reads a requested value from a PLC.
      *
      * @param readRequest object describing the type and location of the value.
+     * @param <T>         type that is being requested.
      * @return a {@link CompletableFuture} giving async access to the returned value.
      */
-    CompletableFuture<PlcReadResponse> read(PlcReadRequest readRequest);
+    <T> CompletableFuture<PlcReadResponse<T>> read(PlcReadRequest<T> readRequest);
 
 }
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcWriter.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcWriter.java
index d427dc8..53add5b 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcWriter.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcWriter.java
@@ -33,8 +33,9 @@
      * Writes a given value to a PLC.
      *
      * @param writeRequest object describing the type, location and value that whould be written.
+     * @param <T>          type that is being requested.
      * @return a {@link CompletableFuture} giving async access to the response of the write operation.
      */
-    CompletableFuture<PlcWriteResponse> write(PlcWriteRequest writeRequest);
+    <T> CompletableFuture<PlcWriteResponse<T>> write(PlcWriteRequest<T> writeRequest);
 
 }
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcMessage.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcMessage.java
index aed337d..d720136 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcMessage.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcMessage.java
@@ -20,7 +20,9 @@
 
 /**
  * Base type for all forms of messages passed between plc and plc4x system.
+ * @param <T>         type that is being handled.
  */
-public interface PlcMessage {
+@SuppressWarnings("unused")
+public interface PlcMessage<T> {
 
 }
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadRequest.java
index 5c3e522..dbf8d59 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadRequest.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadRequest.java
@@ -24,33 +24,33 @@
 import java.util.LinkedList;
 import java.util.List;
 
-public class PlcReadRequest implements PlcRequest {
+public class PlcReadRequest<T> implements PlcRequest<T> {
 
-    private final List<ReadRequestItem> readRequestItems;
+    private final List<ReadRequestItem<T>> readRequestItems;
 
     public PlcReadRequest() {
         this.readRequestItems = new LinkedList<>();
     }
 
-    public PlcReadRequest(Class dataType, Address address) {
+    public PlcReadRequest(Class<T> dataType, Address address) {
         this();
-        addItem(new ReadRequestItem(dataType, address));
+        addItem(new ReadRequestItem<>(dataType, address));
     }
 
-    public PlcReadRequest(Class dataType, Address address, int size) {
+    public PlcReadRequest(Class<T> dataType, Address address, int size) {
         this();
-        addItem(new ReadRequestItem(dataType, address, size));
+        addItem(new ReadRequestItem<>(dataType, address, size));
     }
 
-    public PlcReadRequest(List<ReadRequestItem> readRequestItems) {
+    public PlcReadRequest(List<ReadRequestItem<T>> readRequestItems) {
         this.readRequestItems = readRequestItems;
     }
 
-    public void addItem(ReadRequestItem readRequestItem) {
+    public void addItem(ReadRequestItem<T> readRequestItem) {
         readRequestItems.add(readRequestItem);
     }
 
-    public List<ReadRequestItem> getReadRequestItems() {
+    public List<ReadRequestItem<T>> getReadRequestItems() {
         return readRequestItems;
     }
 
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadResponse.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadResponse.java
index 8d4d7d9..7b31cb3 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadResponse.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcReadResponse.java
@@ -22,12 +22,12 @@
 
 import java.util.List;
 
-public class PlcReadResponse implements PlcResponse {
+public class PlcReadResponse<T> implements PlcResponse {
 
     private final PlcReadRequest request;
-    private final List<ReadResponseItem> responseItems;
+    private final List<ReadResponseItem<T>> responseItems;
 
-    public PlcReadResponse(PlcReadRequest request, List<ReadResponseItem> responseItems) {
+    public PlcReadResponse(PlcReadRequest request, List<ReadResponseItem<T>> responseItems) {
         this.request = request;
         this.responseItems = responseItems;
     }
@@ -36,7 +36,7 @@
         return request;
     }
 
-    public List<ReadResponseItem> getResponseItems() {
+    public List<ReadResponseItem<T>> getResponseItems() {
         return responseItems;
     }
 
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcRequest.java
index f2367f9..40a4b75 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcRequest.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcRequest.java
@@ -21,6 +21,6 @@
 /**
  * Base type for all messages sent from the plc4x system to a connected plc.
  */
-public interface PlcRequest extends PlcMessage {
+public interface PlcRequest<T> extends PlcMessage<T> {
 
 }
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteRequest.java
index ecfdf53..c5cfd93 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteRequest.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteRequest.java
@@ -24,33 +24,33 @@
 import java.util.LinkedList;
 import java.util.List;
 
-public class PlcWriteRequest implements PlcRequest {
+public class PlcWriteRequest<T> implements PlcRequest<T> {
 
-    private final List<WriteRequestItem> requestItems;
+    private final List<WriteRequestItem<T>> requestItems;
 
     public PlcWriteRequest() {
         this.requestItems = new LinkedList<>();
     }
 
-    public PlcWriteRequest(Class dataType, Address address, Object value) {
+    public PlcWriteRequest(Class<T> dataType, Address address, T value) {
         this();
-        addItem(new WriteRequestItem(dataType, address, value));
+        addItem(new WriteRequestItem<>(dataType, address, value));
     }
 
-    public PlcWriteRequest(Class dataType, Address address, Object[] values) {
+    public PlcWriteRequest(Class<T> dataType, Address address, T[] values) {
         this();
-        addItem(new WriteRequestItem(dataType, address, values));
+        addItem(new WriteRequestItem<>(dataType, address, values));
     }
 
-    public PlcWriteRequest(List<WriteRequestItem> requestItems) {
+    public PlcWriteRequest(List<WriteRequestItem<T>> requestItems) {
         this.requestItems = requestItems;
     }
 
-    public void addItem(WriteRequestItem requestItem) {
+    public void addItem(WriteRequestItem<T> requestItem) {
         requestItems.add(requestItem);
     }
 
-    public List<WriteRequestItem> getRequestItems() {
+    public List<WriteRequestItem<T>> getRequestItems() {
         return requestItems;
     }
 
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteResponse.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteResponse.java
index 9e8a925..58b8ce7 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteResponse.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcWriteResponse.java
@@ -22,21 +22,21 @@
 
 import java.util.List;
 
-public class PlcWriteResponse implements PlcResponse {
+public class PlcWriteResponse<T> implements PlcResponse {
 
-    private final PlcWriteRequest request;
-    private final List<WriteResponseItem> responseItems;
+    private final PlcWriteRequest<T> request;
+    private final List<WriteResponseItem<T>> responseItems;
 
-    public PlcWriteResponse(PlcWriteRequest request, List<WriteResponseItem> responseItems) {
+    public PlcWriteResponse(PlcWriteRequest<T> request, List<WriteResponseItem<T>> responseItems) {
         this.request = request;
         this.responseItems = responseItems;
     }
 
-    public PlcWriteRequest getRequest() {
+    public PlcWriteRequest<T> getRequest() {
         return request;
     }
 
-    public List<WriteResponseItem> getResponseItems() {
+    public List<WriteResponseItem<T>> getResponseItems() {
         return responseItems;
     }
 
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/UncheckedPlcReadRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/UncheckedPlcReadRequest.java
new file mode 100644
index 0000000..a60995f
--- /dev/null
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/UncheckedPlcReadRequest.java
@@ -0,0 +1,48 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.plc4x.java.api.messages;
+
+import org.apache.plc4x.java.api.messages.items.ReadRequestItem;
+import org.apache.plc4x.java.api.model.Address;
+
+import java.util.List;
+
+@SuppressWarnings("unchecked")
+public class UncheckedPlcReadRequest extends PlcReadRequest<Object> {
+
+    public UncheckedPlcReadRequest() {
+    }
+
+    public UncheckedPlcReadRequest(Class dataType, Address address) {
+        super(dataType, address);
+    }
+
+    public UncheckedPlcReadRequest(Class dataType, Address address, int size) {
+        super(dataType, address, size);
+    }
+
+    public UncheckedPlcReadRequest(List list) {
+        super(list);
+    }
+
+    @Override
+    public void addItem(ReadRequestItem readRequestItem) {
+        super.addItem(readRequestItem);
+    }
+}
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/UncheckedPlcWriteRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/UncheckedPlcWriteRequest.java
new file mode 100644
index 0000000..83ac035
--- /dev/null
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/UncheckedPlcWriteRequest.java
@@ -0,0 +1,48 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.plc4x.java.api.messages;
+
+import org.apache.plc4x.java.api.messages.items.WriteRequestItem;
+import org.apache.plc4x.java.api.model.Address;
+
+import java.util.List;
+
+@SuppressWarnings("unchecked")
+public class UncheckedPlcWriteRequest extends PlcWriteRequest<Object> {
+
+    public UncheckedPlcWriteRequest() {
+    }
+
+    public UncheckedPlcWriteRequest(Class<Object> dataType, Address address, Object value) {
+        super(dataType, address, value);
+    }
+
+    public UncheckedPlcWriteRequest(Class<Object> dataType, Address address, Object[] values) {
+        super(dataType, address, values);
+    }
+
+    public UncheckedPlcWriteRequest(List<WriteRequestItem<Object>> writeRequestItems) {
+        super(writeRequestItems);
+    }
+
+    @Override
+    public void addItem(WriteRequestItem requestItem) {
+        super.addItem(requestItem);
+    }
+}
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadRequestItem.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadRequestItem.java
index b4e29fb..6086959 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadRequestItem.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadRequestItem.java
@@ -20,27 +20,27 @@
 
 import org.apache.plc4x.java.api.model.Address;
 
-public class ReadRequestItem {
+public class ReadRequestItem<T> {
 
-    private final Class datatype;
+    private final Class<T> datatype;
 
     private final Address address;
 
     private final int size;
 
-    public ReadRequestItem(Class datatype, Address address) {
+    public ReadRequestItem(Class<T> datatype, Address address) {
         this.datatype = datatype;
         this.address = address;
         this.size = 1;
     }
 
-    public ReadRequestItem(Class datatype, Address address, int size) {
+    public ReadRequestItem(Class<T> datatype, Address address, int size) {
         this.datatype = datatype;
         this.address = address;
         this.size = size;
     }
 
-    public Class getDatatype() {
+    public Class<T> getDatatype() {
         return datatype;
     }
 
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadResponseItem.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadResponseItem.java
index 7bf33bb..286fb64 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadResponseItem.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/ReadResponseItem.java
@@ -22,15 +22,15 @@
 
 import java.util.List;
 
-public class ReadResponseItem {
+public class ReadResponseItem<T> {
 
-    private final ReadRequestItem requestItem;
+    private final ReadRequestItem<T> requestItem;
 
     private final ResponseCode responseCode;
 
-    private final List<Object> values;
+    private final List<T> values;
 
-    public ReadResponseItem(ReadRequestItem requestItem, ResponseCode responseCode, List<Object> values) {
+    public ReadResponseItem(ReadRequestItem<T> requestItem, ResponseCode responseCode, List<T> values) {
         this.requestItem = requestItem;
         this.responseCode = responseCode;
         this.values = values;
@@ -44,7 +44,7 @@
         return responseCode;
     }
 
-    public List<Object> getValues() {
+    public List<T> getValues() {
         return values;
     }
 
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteRequestItem.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteRequestItem.java
index c482359..44111ae 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteRequestItem.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteRequestItem.java
@@ -20,27 +20,31 @@
 
 import org.apache.plc4x.java.api.model.Address;
 
-public class WriteRequestItem {
+import java.lang.reflect.Array;
 
-    private final Class datatype;
+public class WriteRequestItem<T> {
+
+    private final Class<T> datatype;
 
     private final Address address;
 
-    private final Object[] values;
+    private final T[] values;
 
-    public WriteRequestItem(Class datatype, Address address, Object value) {
+    @SuppressWarnings("unchecked")
+    public WriteRequestItem(Class<T> datatype, Address address, T value) {
         this.datatype = datatype;
         this.address = address;
-        this.values = new Object[]{value};
+        this.values = (T[]) Array.newInstance(datatype, 1);
+        this.values[0] = value;
     }
 
-    public WriteRequestItem(Class datatype, Address address, Object[] values) {
+    public WriteRequestItem(Class<T> datatype, Address address, T[] values) {
         this.datatype = datatype;
         this.address = address;
         this.values = values;
     }
 
-    public Class getDatatype() {
+    public Class<T> getDatatype() {
         return datatype;
     }
 
@@ -48,7 +52,7 @@
         return address;
     }
 
-    public Object[] getValues() {
+    public T[] getValues() {
         return values;
     }
 
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteResponseItem.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteResponseItem.java
index 5761c2b..524f68f 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteResponseItem.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/items/WriteResponseItem.java
@@ -20,18 +20,18 @@
 
 import org.apache.plc4x.java.api.types.ResponseCode;
 
-public class WriteResponseItem {
+public class WriteResponseItem<T> {
 
-    private final WriteRequestItem requestItem;
+    private final WriteRequestItem<T> requestItem;
 
     private final ResponseCode responseCode;
 
-    public WriteResponseItem(WriteRequestItem requestItem, ResponseCode responseCode) {
+    public WriteResponseItem(WriteRequestItem<T> requestItem, ResponseCode responseCode) {
         this.requestItem = requestItem;
         this.responseCode = responseCode;
     }
 
-    public WriteRequestItem getRequestItem() {
+    public WriteRequestItem<T> getRequestItem() {
         return requestItem;
     }
 
diff --git a/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/Plc4XS7Protocol.java b/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/Plc4XS7Protocol.java
index d234a46..c9b44c2 100644
--- a/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/Plc4XS7Protocol.java
+++ b/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/Plc4XS7Protocol.java
@@ -50,8 +50,6 @@
 
 public class Plc4XS7Protocol extends MessageToMessageCodec<S7Message, PlcRequestContainer> {
 
-    private static final Logger logger = LoggerFactory.getLogger(S7PlcConnection.class);
-
     private static final AtomicInteger tpduGenerator = new AtomicInteger(1);
 
     private Map<Short, PlcRequestContainer> requests;
@@ -65,7 +63,7 @@
         if (msg.getRequest() instanceof PlcReadRequest) {
             List<VarParameterItem> parameterItems = new LinkedList<>();
 
-            PlcReadRequest readRequest = (PlcReadRequest) msg.getRequest();
+            PlcReadRequest<?> readRequest = (PlcReadRequest<?>) msg.getRequest();
             for (ReadRequestItem requestItem : readRequest.getReadRequestItems()) {
                 // Try to get the correct S7 transport size for the given data type.
                 // (Map PLC4X data type to S7 data type)
@@ -78,7 +76,7 @@
                 VarParameterItem varParameterItem = encodeVarParameterItem(requestItem.getAddress(), transportSize, requestItem.getSize());
                 parameterItems.add(varParameterItem);
             }
-            VarParameter readVarParameter =  new VarParameter(ParameterType.READ_VAR, parameterItems);
+            VarParameter readVarParameter = new VarParameter(ParameterType.READ_VAR, parameterItems);
 
             // Assemble the request.
             S7RequestMessage s7ReadRequest = new S7RequestMessage(MessageType.JOB,
@@ -88,11 +86,11 @@
             requests.put(s7ReadRequest.getTpduReference(), msg);
 
             out.add(s7ReadRequest);
-        } else if(msg.getRequest() instanceof PlcWriteRequest) {
+        } else if (msg.getRequest() instanceof PlcWriteRequest) {
             List<VarParameterItem> parameterItems = new LinkedList<>();
             List<VarPayloadItem> payloadItems = new LinkedList<>();
 
-            PlcWriteRequest writeRequest = (PlcWriteRequest) msg.getRequest();
+            PlcWriteRequest<?> writeRequest = (PlcWriteRequest<?>) msg.getRequest();
             for (WriteRequestItem requestItem : writeRequest.getRequestItems()) {
                 // Try to get the correct S7 transport size for the given data type.
                 // (Map PLC4X data type to S7 data type)
@@ -116,7 +114,7 @@
 
                 payloadItems.add(varPayloadItem);
             }
-            VarParameter writeVarParameter =  new VarParameter(ParameterType.WRITE_VAR, parameterItems);
+            VarParameter writeVarParameter = new VarParameter(ParameterType.WRITE_VAR, parameterItems);
             VarPayload writeVarPayload = new VarPayload(ParameterType.WRITE_VAR, payloadItems);
 
             // Assemble the request.
@@ -130,9 +128,10 @@
         }
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     protected void decode(ChannelHandlerContext ctx, S7Message msg, List<Object> out) throws Exception {
-        if(msg instanceof S7ResponseMessage) {
+        if (msg instanceof S7ResponseMessage) {
             S7ResponseMessage responseMessage = (S7ResponseMessage) msg;
             short tpduReference = responseMessage.getTpduReference();
             if (requests.containsKey(tpduReference)) {
@@ -142,14 +141,14 @@
 
                 // Handle the response to a read request.
                 if (requestContainer.getRequest() instanceof PlcReadRequest) {
-                    PlcReadRequest plcReadRequest = (PlcReadRequest) requestContainer.getRequest();
+                    PlcReadRequest<?> plcReadRequest = (PlcReadRequest<?>) requestContainer.getRequest();
 
                     List<ReadResponseItem> responseItems = new LinkedList<>();
                     VarPayload payload = responseMessage.getPayload(VarPayload.class);
                     // If the numbers of items don't match, we're in big trouble as the only
                     // way to know how to interpret the responses is by aligning them with the
                     // items from the request as this information is not returned by the PLC.
-                    if(plcReadRequest.getReadRequestItems().size() != payload.getPayloadItems().size()) {
+                    if (plcReadRequest.getReadRequestItems().size() != payload.getPayloadItems().size()) {
                         throw new PlcProtocolException(
                             "The number of requested items doesn't match the number of returned items");
                     }
@@ -158,20 +157,20 @@
                         VarPayloadItem payloadItem = payloadItems.get(i);
 
                         // Get the request item for this payload item
-                        ReadRequestItem requestItem = plcReadRequest.getReadRequestItems().get(i);
+                        ReadRequestItem<?> requestItem = plcReadRequest.getReadRequestItems().get(i);
 
                         ResponseCode responseCode = decodeResponseCode(payloadItem.getReturnCode());
 
-                        ReadResponseItem responseItem;
+                        ReadResponseItem<?> responseItem;
                         // Something went wrong.
-                        if(responseCode != ResponseCode.OK) {
-                            responseItem = new ReadResponseItem(requestItem, responseCode, null);
+                        if (responseCode != ResponseCode.OK) {
+                            responseItem = new ReadResponseItem<>(requestItem, responseCode, null);
                         }
                         // All Ok.
                         else {
                             byte[] data = payloadItem.getData();
                             Class<?> datatype = requestItem.getDatatype();
-                            List<Object> value = decodeData(datatype, data);
+                            List<?> value = decodeData(datatype, data);
                             responseItem = new ReadResponseItem(requestItem, responseCode, value);
                         }
                         responseItems.add(responseItem);
@@ -181,13 +180,13 @@
 
                 // Handle the response to a write request.
                 else if (requestContainer.getRequest() instanceof PlcWriteRequest) {
-                    PlcWriteRequest plcWriteRequest = (PlcWriteRequest) requestContainer.getRequest();
+                    PlcWriteRequest<?> plcWriteRequest = (PlcWriteRequest) requestContainer.getRequest();
                     List<WriteResponseItem> responseItems = new LinkedList<>();
                     VarPayload payload = responseMessage.getPayload(VarPayload.class);
                     // If the numbers of items don't match, we're in big trouble as the only
                     // way to know how to interpret the responses is by aligning them with the
                     // items from the request as this information is not returned by the PLC.
-                    if(plcWriteRequest.getRequestItems().size() != payload.getPayloadItems().size()) {
+                    if (plcWriteRequest.getRequestItems().size() != payload.getPayloadItems().size()) {
                         throw new PlcProtocolException(
                             "The number of requested items doesn't match the number of returned items");
                     }
@@ -286,7 +285,7 @@
     }
 
     private byte[] encodeData(Object[] values) {
-        if(values.length == 0) {
+        if (values.length == 0) {
             return null;
         }
         byte[] result = null;
@@ -294,24 +293,24 @@
         if (valueType == Boolean.class) {
             // TODO: Check if this is true and the result is not Math.ceil(values.lenght / 8)
             result = new byte[values.length];
-            for(int i = 0; i < values.length; i++) {
+            for (int i = 0; i < values.length; i++) {
                 result[i] = (byte) (((Boolean) values[i]) ? 0x01 : 0x00);
             }
         } else if (valueType == Byte[].class) {
             result = new byte[values.length];
-            for(int i = 0; i < values.length; i++) {
+            for (int i = 0; i < values.length; i++) {
                 result[i] = (byte) values[i];
             }
         } else if (valueType == Short.class) {
             result = new byte[values.length * 2];
-            for(int i = 0; i < values.length; i++) {
+            for (int i = 0; i < values.length; i++) {
                 short intValue = (short) values[i];
                 result[i * 2] = (byte) ((intValue & 0xff00) >> 8);
                 result[(i * 2) + 1] = (byte) (intValue & 0xff);
             }
         } else if (valueType == Integer.class) {
             result = new byte[values.length * 4];
-            for(int i = 0; i < values.length; i++) {
+            for (int i = 0; i < values.length; i++) {
                 int intValue = (int) values[i];
                 result[i * 4] = (byte) ((intValue & 0xff000000) >> 24);
                 result[(i * 4) + 1] = (byte) ((intValue & 0x00ff0000) >> 16);
@@ -322,7 +321,7 @@
             result = null;
         } else if (valueType == Float.class) {
             result = new byte[values.length * 4];
-            for(int i = 0; i < values.length; i++) {
+            for (int i = 0; i < values.length; i++) {
                 float floatValue = (float) values[i];
                 int intValue = Float.floatToIntBits(floatValue);
                 result[i * 4] = (byte) ((intValue & 0xff000000) >> 24);
@@ -341,7 +340,7 @@
     ////////////////////////////////////////////////////////////////////////////////
 
     private ResponseCode decodeResponseCode(DataTransportErrorCode dataTransportErrorCode) {
-        if(dataTransportErrorCode != null) {
+        if (dataTransportErrorCode != null) {
             switch (dataTransportErrorCode) {
                 case OK:
                     return ResponseCode.OK;
@@ -354,25 +353,26 @@
         return ResponseCode.INTERNAL_ERROR;
     }
 
-    private List<Object> decodeData(Class<?> datatype, byte[] s7Data) throws PlcProtocolException {
-        if(s7Data.length == 0) {
+    @SuppressWarnings("unchecked")
+    private <T> List<T> decodeData(Class<T> datatype, byte[] s7Data) throws PlcProtocolException {
+        if (s7Data.length == 0) {
             return null;
         }
         List<Object> result = new LinkedList<>();
-        for(int i = 0; i < s7Data.length;) {
+        for (int i = 0; i < s7Data.length;) {
             if (datatype == Boolean.class) {
                 result.add((s7Data[i] & 0x01) == 0x01);
-                i+=1;
+                i += 1;
             } else if (datatype == Byte.class) {
                 result.add(s7Data[i]);
-                i+=1;
+                i += 1;
             } else if (datatype == Short.class) {
-                result.add((short) (((s7Data[i] & 0xff) << 8) | (s7Data[i+1] & 0xff)));
-                i+=2;
+                result.add((short) (((s7Data[i] & 0xff) << 8) | (s7Data[i + 1] & 0xff)));
+                i += 2;
             } else if (datatype == Integer.class) {
                 result.add((((s7Data[i] & 0xff) << 24) | ((s7Data[i + 1] & 0xff) << 16) |
                     ((s7Data[i + 2] & 0xff) << 8) | (s7Data[i + 3] & 0xff)));
-                i+=4;
+                i += 4;
             } else if (datatype == Float.class) {
                 // Description of the Real number format:
                 // https://www.sps-lehrgang.de/zahlenformate-step7/#c144
@@ -380,12 +380,12 @@
                 int intValue = (((s7Data[i] & 0xff) << 24) | ((s7Data[i + 1] & 0xff) << 16) |
                     ((s7Data[i + 2] & 0xff) << 8) | (s7Data[i + 3] & 0xff));
                 result.add(Float.intBitsToFloat(intValue));
-                i+=4;
+                i += 4;
             } else {
                 throw new PlcProtocolException("Unsupported datatype " + datatype.getSimpleName());
             }
         }
-        return result;
+        return (List<T>) result;
     }
 
 }
diff --git a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcReaderSample.java b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcReaderSample.java
index 9415179..864c9f3 100644
--- a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcReaderSample.java
+++ b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcReaderSample.java
@@ -22,9 +22,9 @@
 import org.apache.plc4x.java.api.connection.PlcConnection;
 import org.apache.plc4x.java.api.connection.PlcReader;
 import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
 import org.apache.plc4x.java.api.messages.items.ReadResponseItem;
 import org.apache.plc4x.java.api.model.Address;
-import org.apache.plc4x.java.api.messages.PlcReadResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,7 +44,7 @@
     public static void main(String[] args) throws Exception {
         // Create a connection to the S7 PLC (s7://{hostname/ip}/{racknumber}/{slotnumber})
         logger.info("Connecting");
-        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")){
+        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")) {
             logger.info("Connected");
 
             Optional<PlcReader> reader = plcConnection.getReader();
@@ -63,8 +63,8 @@
                 // Read synchronously ...
                 // NOTICE: the ".get()" immediately lets this thread pause till
                 // the response is processed and available.
-                PlcReadResponse plcReadResponse = plcReader.read(new PlcReadRequest(Byte.class, inputs)).get();
-                List<ReadResponseItem> responseItems = plcReadResponse.getResponseItems();
+                PlcReadResponse<Byte> plcReadResponse = plcReader.read(new PlcReadRequest<>(Byte.class, inputs)).get();
+                List<ReadResponseItem<Byte>> responseItems = plcReadResponse.getResponseItems();
                 System.out.println("Inputs: " + responseItems.get(0).getValues().get(0));
 
                 //////////////////////////////////////////////////////////
diff --git a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcScanner.java b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcScanner.java
index fdc7fac..4f072e6 100644
--- a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcScanner.java
+++ b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcScanner.java
@@ -43,7 +43,7 @@
     public static void main(String[] args) throws Exception {
         // Create a connection to the S7 PLC (s7://{hostname/ip}/{racknumber}/{slotnumber})
         logger.info("Connecting");
-        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")){
+        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")) {
             logger.info("Connected");
 
             Optional<PlcReader> reader = plcConnection.getReader();
@@ -54,18 +54,18 @@
                 for (MemoryArea memoryArea : MemoryArea.values()) {
                     System.out.println(memoryArea);
                     System.out.println("------------------------------------------");
-                    for(int i = 0; i < 8959; i++) {
+                    for (int i = 0; i < 8959; i++) {
                         try {
                             Address address;
-                            if(memoryArea == MemoryArea.DATA_BLOCKS) {
+                            if (memoryArea == MemoryArea.DATA_BLOCKS) {
                                 address = plcConnection.parseAddress("DATA_BLOCKS/1/" + i);
                             } else {
                                 address = plcConnection.parseAddress(memoryArea.name() + "/" + i);
                             }
-                            PlcReadResponse plcReadResponse = plcReader.read(
-                                new PlcReadRequest(Byte.class, address)).get();
-                            Byte data = (Byte) plcReadResponse.getResponseItems().get(0).getValues().get(0);
-                            if(data != null && data != 0) {
+                            PlcReadResponse<Byte> plcReadResponse = plcReader.read(
+                                new PlcReadRequest<>(Byte.class, address)).get();
+                            Byte data = plcReadResponse.getResponseItems().get(0).getValues().get(0);
+                            if (data != null && data != 0) {
                                 System.out.println(String.format(
                                     "Response: Memory Area: %s Index: %d Value: %02X", memoryArea.name(), i, data));
                             }
diff --git a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcTestConsole.java b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcTestConsole.java
index 58cc851..96e155e 100644
--- a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcTestConsole.java
+++ b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcTestConsole.java
@@ -44,7 +44,7 @@
     public static void main(String[] args) throws Exception {
         // Create a connection to the S7 PLC (s7://{hostname/ip}/{racknumber}/{slotnumber})
         logger.info("Connecting");
-        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")){
+        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")) {
             logger.info("Connected");
 
             Optional<PlcReader> reader = plcConnection.getReader();
@@ -54,13 +54,13 @@
 
                 Scanner scanner = new Scanner(System.in);
                 String line;
-                while(!"exit".equalsIgnoreCase(line = scanner.next())) {
+                while (!"exit".equalsIgnoreCase(line = scanner.next())) {
                     try {
                         Address address = plcConnection.parseAddress(line);
-                        PlcReadResponse plcReadResponse = plcReader.read(new PlcReadRequest(Byte.class, address)).get();
-                        List<Object> data = plcReadResponse.getResponseItems().get(0).getValues();
+                        PlcReadResponse<Byte> plcReadResponse = plcReader.read(new PlcReadRequest<>(Byte.class, address)).get();
+                        List<Byte> data = plcReadResponse.getResponseItems().get(0).getValues();
                         System.out.println("Response: " + data.get(0));
-                    } catch(Exception e) {
+                    } catch (Exception e) {
                         e.printStackTrace();
                     }
                 }
diff --git a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcWriterSample.java b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcWriterSample.java
index 3edc52a..159e588 100644
--- a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcWriterSample.java
+++ b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/S7PlcWriterSample.java
@@ -21,7 +21,8 @@
 import org.apache.plc4x.java.PlcDriverManager;
 import org.apache.plc4x.java.api.connection.PlcConnection;
 import org.apache.plc4x.java.api.connection.PlcWriter;
-import org.apache.plc4x.java.api.messages.*;
+import org.apache.plc4x.java.api.messages.PlcWriteRequest;
+import org.apache.plc4x.java.api.messages.PlcWriteResponse;
 import org.apache.plc4x.java.api.model.Address;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,7 +42,7 @@
     public static void main(String[] args) throws Exception {
         // Create a connection to the S7 PLC (s7://{hostname/ip}/{racknumber}/{slotnumber})
         logger.info("Connecting");
-        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")){
+        try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://192.168.0.1/0/0")) {
             logger.info("Connected");
 
             Optional<PlcWriter> writer = plcConnection.getWriter();
@@ -53,8 +54,8 @@
                 // Write synchronously ...
                 // NOTICE: the ".get()" immediately lets this thread pause till
                 // the response is processed and available.
-                PlcWriteResponse plcWriteResponse = plcWriter.write(
-                    new PlcWriteRequest(Float.class, inputs, 2.0f)).get();
+                PlcWriteResponse<Float> plcWriteResponse = plcWriter.write(
+                    new PlcWriteRequest<>(Float.class, inputs, 2.0f)).get();
                 System.out.println("Written: " + plcWriteResponse.getResponseItems().get(0).getResponseCode().name());
             }
         }
diff --git a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/netty/Plc4XS7ProtocolTest.java b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/netty/Plc4XS7ProtocolTest.java
index adafac3..4556f10 100644
--- a/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/netty/Plc4XS7ProtocolTest.java
+++ b/plc4j/protocols/s7/src/test/java/org/apache/plc4x/java/s7/netty/Plc4XS7ProtocolTest.java
@@ -67,7 +67,7 @@
         // Read Request Tests
         {
             LinkedList<Object> out = new LinkedList<>();
-            SUT.encode(null, createMockedContainer(new PlcReadRequest(type, address)), out);
+            SUT.encode(null, createMockedContainer(new PlcReadRequest<>(type, address)), out);
             // TODO: finish the asserts
             assertThat(out).hasSize(1);
         }