validate begin frame type for ws client (#1332)

aklivity · Nov 19, 2024 · 3695c9f · 3695c9f
1 parent 68ce04e
commit 3695c9f
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/...s/src/main/java/io/aklivity/zilla/runtime/binding/ws/internal/stream/WsClientFactory.java b/...s/src/main/java/io/aklivity/zilla/runtime/binding/ws/internal/stream/WsClientFactory.java
@@ -55,6 +55,7 @@
 import io.aklivity.zilla.runtime.binding.ws.internal.types.stream.ChallengeFW;
 import io.aklivity.zilla.runtime.binding.ws.internal.types.stream.DataFW;
 import io.aklivity.zilla.runtime.binding.ws.internal.types.stream.EndFW;
+import io.aklivity.zilla.runtime.binding.ws.internal.types.stream.ExtensionFW;
 import io.aklivity.zilla.runtime.binding.ws.internal.types.stream.FlushFW;
 import io.aklivity.zilla.runtime.binding.ws.internal.types.stream.HttpBeginExFW;
 import io.aklivity.zilla.runtime.binding.ws.internal.types.stream.ResetFW;
@@ -113,6 +114,7 @@ public final class WsClientFactory implements WsStreamFactory
     private final OctetsFW.Builder payloadRW = new OctetsFW.Builder();
 
     private final OctetsFW payloadRO = new OctetsFW();
+    private final ExtensionFW extensionExRO = new ExtensionFW();
 
     private final HttpBeginExFW httpBeginExRO = new HttpBeginExFW();
     private final HttpBeginExFW.Builder httpBeginExRW = new HttpBeginExFW.Builder();
@@ -184,7 +186,10 @@ public MessageConsumer newStream(
         final long initialId = begin.streamId();
         final long authorization = begin.authorization();
         final OctetsFW extension = begin.extension();
-        final WsBeginExFW wsBeginEx = extension.get(wsBeginExRO::tryWrap);
+        final ExtensionFW extensionEx = extension.get(extensionExRO::tryWrap);
+        final WsBeginExFW wsBeginEx = extensionEx != null && extensionEx.typeId() == wsTypeId
+            ? extension.get(wsBeginExRO::tryWrap)
+            : null;
 
         MessageConsumer newStream = null;