WIP

runtime/binding-tls/NOTICE

@@ -12,6 +12,18 @@ specific language governing permissions and limitations
 under the License.
 
 This project includes:
+  agrona under The Apache License, Version 2.0
+  ICU4J under Unicode/ICU License
+  Jackson-annotations under The Apache Software License, Version 2.0
+  Jackson-core under The Apache Software License, Version 2.0
+  jackson-databind under The Apache Software License, Version 2.0
+  Jackson-dataformat-YAML under The Apache Software License, Version 2.0
+  Jakarta JSON Processing API under Eclipse Public License 2.0 or GNU General Public License, version 2 with the GNU Classpath Exception
+  JSON-B API under Eclipse Public License 2.0 or GNU General Public License, version 2 with the GNU Classpath Exception
+  org.leadpony.justify under The Apache Software License, Version 2.0
+  SnakeYAML under Apache License, Version 2.0
+  zilla::runtime::common under The Apache Software License, Version 2.0
+  zilla::runtime::engine under The Apache Software License, Version 2.0
 
 
 This project also includes code under copyright of the following entities:

runtime/binding-tls/pom.xml

@@ -41,7 +41,6 @@
       <groupId>${project.groupId}</groupId>
       <artifactId>engine</artifactId>
       <version>${project.version}</version>
-      <scope>provided</scope>
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -56,6 +55,12 @@
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>binding-echo</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
@@ -241,6 +246,8 @@
             <includes>
               <include>org.agrona:agrona</include>
               <include>io.aklivity.zilla:engine</include>
+              <include>io.aklivity.zilla:common</include>
+              <include>io.aklivity.zilla:binding-echo</include>
               <include>org.openjdk.jmh:jmh-core</include>
               <include>net.sf.jopt-simple:jopt-simple</include>
               <include>org.apache.commons:commons-math3</include>

runtime/binding-tls/src/test/java/io/aklivity/zilla/runtime/binding/tls/internal/bench/TlsHandshakeBM.java

@@ -15,11 +15,12 @@
  */
 package io.aklivity.zilla.runtime.binding.tls.internal.bench;
 
+import static io.aklivity.zilla.runtime.engine.EngineConfiguration.ENGINE_DIRECTORY;
 import static io.aklivity.zilla.runtime.engine.config.KindConfig.CLIENT;
 import static io.aklivity.zilla.runtime.engine.config.KindConfig.SERVER;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
-import java.io.IOException;
+import java.util.Properties;
 
 import org.agrona.MutableDirectBuffer;
 import org.agrona.concurrent.UnsafeBuffer;
@@ -43,12 +44,9 @@
 
 import io.aklivity.zilla.runtime.binding.tls.internal.types.stream.BeginFW;
 import io.aklivity.zilla.runtime.binding.tls.internal.types.stream.WindowFW;
-import io.aklivity.zilla.runtime.engine.Configuration;
-import io.aklivity.zilla.runtime.engine.binding.BindingContext;
-import io.aklivity.zilla.runtime.engine.binding.BindingFactory;
+import io.aklivity.zilla.runtime.engine.EngineConfiguration;
 import io.aklivity.zilla.runtime.engine.binding.BindingHandler;
 import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
-import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 
 @State(Scope.Benchmark)
@@ -65,18 +63,18 @@ public class TlsHandshakeBM
     private final WindowFW.Builder windowRW = new WindowFW.Builder();
     private final MutableDirectBuffer writeBuffer = new UnsafeBuffer(new byte[BUFFER_SIZE]);
 
-    private BindingHandler handler;
-    private Runnable detacher;
+    private BindingHandler streamFactory;
 
     @Setup(Level.Trial)
-    public void init() throws IOException
+    public void init()
     {
-        BindingFactory bindings = BindingFactory.instantiate();
-        BindingContext context = bindings.create("tls", new Configuration())
-                .supply(new TlsWorker());
+        final Properties properties = new Properties();
+        properties.setProperty(ENGINE_DIRECTORY.name(), "target/zilla-bm");
+        final EngineConfiguration config = new EngineConfiguration(properties);
+        TlsWorker worker = new TlsWorker(config);
 
         NamespaceConfig namespace = NamespaceConfig.builder()
-            .name("echo")
+            .name("tls")
             .binding()
                 .name("tls_client0")
                 .type("tls")
@@ -96,19 +94,14 @@ public void init() throws IOException
                 .build()
             .build();
 
-        BindingConfig binding = namespace.bindings.stream()
-                .filter(b -> "tls_client0".equals(b.name))
-                .findFirst()
-                .get();
+        worker.attach(namespace);
 
-        this.handler = context.attach(binding);
-        this.detacher = () -> context.detach(binding);
+        streamFactory = worker.streamFactory();
     }
 
     @TearDown(Level.Trial)
     public void destroy()
     {
-        detacher.run();
     }
 
     @Setup(Level.Iteration)
@@ -122,7 +115,7 @@ public void handshake(
     {
         final BeginFW begin = beginRW.wrap(writeBuffer, 0, writeBuffer.capacity())
             .originId(0L)
-            .routedId(0L)
+            .routedId(4261135416L)
             .streamId(0L)
             .sequence(0L)
             .acknowledge(0L)
@@ -133,13 +126,14 @@ public void handshake(
             .build();
 
         MessageConsumer sender = MessageConsumer.NOOP;
-        MessageConsumer receiver = handler.newStream(begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof(), sender);
+        MessageConsumer receiver = streamFactory.newStream(
+            begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof(), sender);
 
         receiver.accept(begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof());
 
         final WindowFW window = windowRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                 .originId(0L)
-                .routedId(0L)
+                .routedId(4261135416L)
                 .streamId(0L)
                 .sequence(0L)
                 .acknowledge(0L)

runtime/binding-tls/src/test/java/io/aklivity/zilla/runtime/binding/tls/internal/bench/TlsWorker.java

@@ -15,17 +15,27 @@
  */
 package io.aklivity.zilla.runtime.binding.tls.internal.bench;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+
 import java.net.InetAddress;
 import java.net.URL;
 import java.nio.channels.SelectableChannel;
 import java.time.Clock;
 import java.util.function.LongSupplier;
+import java.util.zip.CRC32C;
 
+import org.agrona.DirectBuffer;
 import org.agrona.MutableDirectBuffer;
+import org.agrona.collections.Long2ObjectHashMap;
+import org.agrona.collections.Object2ObjectHashMap;
 import org.agrona.concurrent.UnsafeBuffer;
 
+import io.aklivity.zilla.runtime.engine.Configuration;
 import io.aklivity.zilla.runtime.engine.EngineConfiguration;
 import io.aklivity.zilla.runtime.engine.EngineContext;
+import io.aklivity.zilla.runtime.engine.binding.Binding;
+import io.aklivity.zilla.runtime.engine.binding.BindingContext;
+import io.aklivity.zilla.runtime.engine.binding.BindingFactory;
 import io.aklivity.zilla.runtime.engine.binding.BindingHandler;
 import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
 import io.aklivity.zilla.runtime.engine.binding.function.MessageReader;
@@ -40,6 +50,7 @@
 import io.aklivity.zilla.runtime.engine.event.EventFormatter;
 import io.aklivity.zilla.runtime.engine.guard.GuardHandler;
 import io.aklivity.zilla.runtime.engine.internal.layouts.BufferPoolLayout;
+import io.aklivity.zilla.runtime.engine.internal.types.stream.FrameFW;
 import io.aklivity.zilla.runtime.engine.metrics.Metric;
 import io.aklivity.zilla.runtime.engine.model.ConverterHandler;
 import io.aklivity.zilla.runtime.engine.model.ValidatorHandler;
@@ -50,16 +61,30 @@ public class TlsWorker implements EngineContext
 {
     private static final int BUFFER_SIZE = 1024 * 8;
     private final MutableDirectBuffer writeBuffer = new UnsafeBuffer(new byte[BUFFER_SIZE]);
+    private final BufferPool bufferPool;
+    private final Long2ObjectHashMap<BindingHandler> handlers;
+    private final Object2ObjectHashMap<String, Binding> bindings;
+    private final BindingFactory factory;
+    private final Configuration config;
+
+    private final FrameFW frameRO = new FrameFW();
+    private final CRC32C crc32C = new CRC32C();
 
     public TlsWorker(
         EngineConfiguration config)
     {
-        final BufferPoolLayout bufferPoolLayout = new BufferPoolLayout.Builder()
-                .path(config.directory().resolve(String.format("buffers%d", index)))
+        this.config = config;
+        this.bufferPool = new BufferPoolLayout.Builder()
+                .path(config.directory().resolve(String.format("buffers%d", 0)))
                 .slotCapacity(config.bufferSlotCapacity())
                 .slotCount(config.bufferPoolCapacity() / config.bufferSlotCapacity())
-                .readonly(readonly)
-                .build();
+                .readonly(false)
+                .build()
+                .bufferPool();
+
+        this.factory = BindingFactory.instantiate();
+        this.bindings = new Object2ObjectHashMap<>();
+        this.handlers = new Long2ObjectHashMap<>();
     }
 
     @Override
@@ -141,28 +166,25 @@ public EventFormatter supplyEventFormatter()
     }
 
     @Override
-    public void attachComposite(NamespaceConfig composite)
+    public void attachComposite(
+        NamespaceConfig composite)
     {
-
     }
 
     @Override
     public void detachComposite(
         NamespaceConfig composite)
     {
-
     }
 
     @Override
     public void detachSender(long replyId)
     {
-
     }
 
     @Override
     public void detachStreams(long bindingId)
     {
-
     }
 
     @Override
@@ -186,7 +208,7 @@ public MutableDirectBuffer writeBuffer()
     @Override
     public BufferPool bufferPool()
     {
-        return null;
+        return bufferPool;
     }
 
     @Override
@@ -279,7 +301,7 @@ public int supplyEventId(
     @Override
     public BindingHandler streamFactory()
     {
-        return null;
+        return this::newStream;
     }
 
     @Override
@@ -369,4 +391,60 @@ public Clock clock()
     {
         return null;
     }
+
+    public void attach(
+        NamespaceConfig namespace)
+    {
+        namespace.bindings.stream()
+            .peek(b -> b.id = crc32c(b.name))
+            .map(b -> b.routes)
+            .forEach(rs -> rs.stream()
+                .peek(r -> r.id = crc32c(r.exit))
+                .forEach(r -> r.authorized = session -> true));
+
+        namespace.bindings.forEach(b ->
+        {
+            Binding binding = supplyBinding(b.type);
+            BindingContext context = binding.supply(this);
+            BindingHandler handler = context.attach(b);
+
+            handlers.put(b.id, handler);
+        });
+    }
+
+    private MessageConsumer newStream(
+        int msgTypeId,
+        DirectBuffer buffer,
+        int index,
+        int length,
+        MessageConsumer sender)
+    {
+        final FrameFW frame = frameRO.wrap(buffer, index, length);
+        final long bindingId = frame.routedId();
+
+        final BindingHandler handler = handlers.get(bindingId);
+
+        return handler.newStream(msgTypeId, buffer, index, length, sender);
+    }
+
+    private Binding supplyBinding(
+        String type)
+    {
+        return bindings.computeIfAbsent(type, this::createBinding);
+    }
+
+    private Binding createBinding(
+        String type)
+    {
+        return factory.create(type, config);
+    }
+
+    private long crc32c(
+        String value)
+    {
+        crc32C.reset();
+        crc32C.update(value.getBytes(UTF_8));
+
+        return crc32C.getValue();
+    }
 }