void subscribeAndWrite(Map<String, Entry<String, Object>> entries) throws Exception {
        PlcConnection connection = new DefaultPlcDriverManager().getConnection("genericcan:virtualcan://");

        Plc4xEmbeddedChannel subscribeChannel = null;
        Plc4xEmbeddedChannel writeChannel = null;
        if (connection instanceof ChannelExposingConnection) {
            Channel channel = ((ChannelExposingConnection) connection).getChannel();
            if (channel instanceof Plc4xEmbeddedChannel) {
                subscribeChannel = (Plc4xEmbeddedChannel) channel;
                writeChannel = (Plc4xEmbeddedChannel) channel;
            }
        }
        if (subscribeChannel == null) {
            throw new IllegalArgumentException("Invalid configuration");
        }

        CountDownLatch latch = new CountDownLatch(1);

        final AtomicReference<PlcSubscriptionEvent> plcEvent = new AtomicReference<>();
        Builder subscriptionRequestBuilder = connection.subscriptionRequestBuilder();
        entries.forEach((k, v) -> {
            subscriptionRequestBuilder.addEventTagAddress(k, v.getKey());
        });

        PlcSubscriptionRequest subscriptionRequest = subscriptionRequestBuilder.build();
        subscriptionRequest.execute().whenComplete((reply, error) -> {
            if (error != null) {
                fail(error);
                return;
            }

            reply.getSubscriptionHandle(entries.keySet().iterator().next()).register(event -> {
                plcEvent.set(event);
                latch.countDown();
            });
        });

        PlcWriteRequest.Builder writeRequestBuilder = connection.writeRequestBuilder();
        entries.forEach((k, v) -> {
            writeRequestBuilder.addTagAddress(k, v.getKey(), v.getValue());
        });
        PlcWriteRequest writeRequest = writeRequestBuilder.build();
        writeRequest.execute().whenComplete((reply, error) -> {
            if (error != null) {
                fail(error);
            }
        }).get();

        // copy outbound message to inbound queue to confirm that transport API works and subscription
        // is properly matched against incoming message
        ByteBuf outgoing = writeChannel.flushOutbound().readOutbound();
        subscribeChannel.writeInbound(outgoing);
        subscribeChannel.flushInbound();

        latch.await();

        PlcSubscriptionEvent event = plcEvent.get();
        entries.forEach((k, v) -> {
            Object object = event.getObject(k);
            if (!v.getValue().getClass().isArray()) {
                assertEquals(v.getValue(), object);
                return;
            }

            // comparing arrays is a bit of nightmare due to primitives
            int length = Array.getLength(v.getValue());
            int readLength = Array.getLength(object);
            if (readLength != length) {
                throw new IllegalArgumentException("Return value length do not match reference value");
            }
            for (int index = 0; index < length; index++) {
                assertEquals(Array.get(v.getValue(), index), Array.get(object, index));
            }
        });

    }
Test of generic can driver with virtual can transport.

This test have additional role of confirming end to end behavior of driver and transport layer.
The virtual can transport rely on netty channel/pipeline infrastructure, hence it does not have
any backend such as memory queue. It simply converts message to a stream and expect it to be read.
plc4j/drivers/can/src/test/java/org/apache/plc4x/java/can/generic/GenericCANDriverTest.java