aklivity · jfallows · Nov 4, 2024 · Oct 17, 2024 · Oct 23, 2024 · Oct 23, 2024
diff --git a/...io/aklivity/zilla/runtime/binding/filesystem/internal/stream/FileSystemServerFactory.java b/...io/aklivity/zilla/runtime/binding/filesystem/internal/stream/FileSystemServerFactory.java
diff --git a/...java/io/aklivity/zilla/runtime/binding/filesystem/internal/stream/FileSystemServerIT.java b/...java/io/aklivity/zilla/runtime/binding/filesystem/internal/stream/FileSystemServerIT.java
@@ -252,4 +252,73 @@ public void shouldReceiveClientReadAbort() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${app}/create.file.payload/client",
+    })
+    public void shouldCreateFilePayloadOnly() throws Exception
+    {
+        Path targetDirectory = Paths.get("target/files").toAbsolutePath();
+        Path indexFile = targetDirectory.resolve("index_write.html");
+
+        Files.deleteIfExists(indexFile);
+
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${app}/write.file.payload.modified/client",
+    })
+    public void shouldWriteFilePayloadModified() throws Exception
+    {
+        Path targetDirectory = Paths.get("target/files").toAbsolutePath();
+        Path indexFile = targetDirectory.resolve("index_write.html");
+
+        Files.createDirectories(targetDirectory);
+
+        Files.write(indexFile, """
+            <html>
+            <head><title>Welcome</title></head>
+            <body>Hello, world</body>
+            </html>
+            """.getBytes());
+
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${app}/write.file.payload.modified.abort/client",
+    })
+    public void shouldWriteFilePayloadModifiedAbort() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${app}/write.file.payload.interrupt/client",
+    })
+    public void shouldWriteFilePayloadInterrupt() throws Exception
+    {
+        Path targetDirectory = Paths.get("target/files").toAbsolutePath();
+        Path indexFile = targetDirectory.resolve("index_write.html");
+
+        Files.createDirectories(targetDirectory);
+
+        Files.write(indexFile, """
+            <html>
+            <head><title>Welcome</title></head>
+            <body>Hello, world</body>
+            </html>
+            """.getBytes());
+
+        k3po.finish();
+    }
 }
diff --git a/...ity/zilla/runtime/binding/http/filesystem/internal/config/HttpFileSystemWithResolver.java b/...ity/zilla/runtime/binding/http/filesystem/internal/config/HttpFileSystemWithResolver.java
@@ -14,8 +14,10 @@
  */
 package io.aklivity.zilla.runtime.binding.http.filesystem.internal.config;
 
+import static io.aklivity.zilla.runtime.binding.http.filesystem.internal.types.FileSystemCapabilities.CREATE_PAYLOAD;
 import static io.aklivity.zilla.runtime.binding.http.filesystem.internal.types.FileSystemCapabilities.READ_EXTENSION;
 import static io.aklivity.zilla.runtime.binding.http.filesystem.internal.types.FileSystemCapabilities.READ_PAYLOAD;
+import static io.aklivity.zilla.runtime.binding.http.filesystem.internal.types.FileSystemCapabilities.WRITE_PAYLOAD;
 
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -32,14 +34,19 @@ public final class HttpFileSystemWithResolver
 {
     public static final int HEADER_METHOD_MASK_HEAD = 1 << READ_EXTENSION.ordinal();
     private static final int HEADER_METHOD_MASK_GET = 1 << READ_PAYLOAD.ordinal() | 1 << READ_EXTENSION.ordinal();
+    public static final int HEADER_METHOD_MASK_POST = 1 << CREATE_PAYLOAD.ordinal();
+    public static final int HEADER_METHOD_MASK_PUT = 1 << WRITE_PAYLOAD.ordinal();
 
     private static final Pattern PARAMS_PATTERN = Pattern.compile("\\$\\{params\\.([a-zA-Z_]+)\\}");
     private static final Pattern PREFER_WAIT_PATTERN = Pattern.compile("wait=(\\d+)");
     private static final String8FW HEADER_METHOD_NAME = new String8FW(":method");
     private static final String8FW HEADER_IF_NONE_MATCH_NAME = new String8FW("if-none-match");
+    private static final String8FW HEADER_IF_MATCH_NAME = new String8FW("if-match");
     private static final String8FW HEADER_PREFER_NAME = new String8FW("prefer");
     private static final String16FW HEADER_METHOD_VALUE_GET = new String16FW("GET");
     private static final String16FW HEADER_METHOD_VALUE_HEAD = new String16FW("HEAD");
+    private static final String16FW HEADER_METHOD_VALUE_POST = new String16FW("POST");
+    private static final String16FW HEADER_METHOD_VALUE_PUT = new String16FW("PUT");
 
     private final String16FW etagRO = new String16FW();
     private final HttpFileSystemWithConfig with;
@@ -73,6 +80,7 @@ public HttpFileSystemWithResult resolve(
             path0 = pathMatcher.replaceAll(replacer);
         }
         String16FW path = new String16FW(path0);
+        String16FW etag = new String16FW("");
 
         HttpHeaderFW method = httpBeginEx.headers().matchFirst(h -> HEADER_METHOD_NAME.equals(h.name()));
         int capabilities = 0;
@@ -86,9 +94,22 @@ else if (HEADER_METHOD_VALUE_GET.equals(method.value()))
             {
                 capabilities = HEADER_METHOD_MASK_GET;
             }
+            else if (HEADER_METHOD_VALUE_POST.equals(method.value()))
+            {
+                capabilities = HEADER_METHOD_MASK_POST;
+            }
+            else if (HEADER_METHOD_VALUE_PUT.equals(method.value()))
+            {
+                capabilities = HEADER_METHOD_MASK_PUT;
+                HttpHeaderFW ifMatched = httpBeginEx.headers().matchFirst(h -> HEADER_IF_MATCH_NAME.equals(h.name()));
+                if (ifMatched != null)
+                {
+                    String16FW value = ifMatched.value();
+                    etag = etagRO.wrap(value.buffer(), value.offset(), value.limit());
+                }
+            }
         }
         HttpHeaderFW ifNotMatched = httpBeginEx.headers().matchFirst(h -> HEADER_IF_NONE_MATCH_NAME.equals(h.name()));
-        String16FW etag = new String16FW("");
         if (ifNotMatched != null)
         {
             String16FW value = ifNotMatched.value();

diff --git a/...ity/zilla/runtime/binding/http/filesystem/internal/stream/HttpFileSystemProxyFactory.java b/...ity/zilla/runtime/binding/http/filesystem/internal/stream/HttpFileSystemProxyFactory.java
@@ -54,13 +54,16 @@ public final class HttpFileSystemProxyFactory implements HttpFileSystemStreamFac
 
     private static final String8FW HEADER_STATUS_NAME = new String8FW(":status");
     private static final String16FW HEADER_STATUS_VALUE_200 = new String16FW("200");
+    private static final String16FW HEADER_STATUS_VALUE_204 = new String16FW("204");
     private static final String16FW HEADER_STATUS_VALUE_304 = new String16FW("304");
     private static final String8FW HEADER_ETAG_NAME = new String8FW("etag");
     private static final String8FW HEADER_CONTENT_TYPE_NAME = new String8FW("content-type");
     private static final String8FW HEADER_CONTENT_LENGTH_NAME = new String8FW("content-length");
     private static final int READ_PAYLOAD_MASK = 1 << FileSystemCapabilities.READ_PAYLOAD.ordinal();
+    private static final int WRITE_PAYLOAD_MASK = 1 << FileSystemCapabilities.WRITE_PAYLOAD.ordinal();
+    private static final int CREATE_PAYLOAD_MASK = 1 << FileSystemCapabilities.CREATE_PAYLOAD.ordinal();
 
-    private static final Predicate<HttpHeaderFW> HEADER_METHOD_GET_OR_HEAD;
+    private static final Predicate<HttpHeaderFW> ALLOWED_HTTP_METHOD;
 
     static
     {
@@ -76,9 +79,23 @@ public final class HttpFileSystemProxyFactory implements HttpFileSystemStreamFac
                 .value("HEAD")
                 .build();
 
+        HttpHeaderFW headerMethodPost = new HttpHeaderFW.Builder()
+            .wrap(new UnsafeBuffer(new byte[512]), 0, 512)
+            .name(":method")
+            .value("POST")
+            .build();
+
+        HttpHeaderFW headerMethodPut = new HttpHeaderFW.Builder()
+            .wrap(new UnsafeBuffer(new byte[512]), 0, 512)
+            .name(":method")
+            .value("PUT")
+            .build();
+
         Predicate<HttpHeaderFW> test = headerMethodGet::equals;
         test = test.or(headerMethodHead::equals);
-        HEADER_METHOD_GET_OR_HEAD = test;
+        test = test.or(headerMethodPost::equals);
+        test = test.or(headerMethodPut::equals);
+        ALLOWED_HTTP_METHOD = test;
     }
 
     private final OctetsFW emptyExRO = new OctetsFW().wrap(new UnsafeBuffer(0L, 0), 0, 0);
@@ -173,7 +190,7 @@ public MessageConsumer newStream(
 
         HttpFileSystemRouteConfig route = null;
 
-        if (binding != null && beginEx.headers().anyMatch(HEADER_METHOD_GET_OR_HEAD))
+        if (binding != null && beginEx.headers().anyMatch(ALLOWED_HTTP_METHOD))
         {
             route = binding.resolve(authorization, beginEx);
         }
@@ -300,6 +317,10 @@ private void onHttpData(
             final long acknowledge = data.acknowledge();
             final long traceId = data.traceId();
             final long authorization = data.authorization();
+            final long budgetId = data.budgetId();
+            final int reserved = data.reserved();
+            final int flags = data.flags();
+            final OctetsFW payload = data.payload();
 
             assert acknowledge <= sequence;
             assert sequence >= initialSeq;
@@ -308,8 +329,15 @@ private void onHttpData(
 
             assert initialAck <= initialSeq;
 
-            doHttpReset(traceId);
-            delegate.doFileSystemAbort(traceId, authorization);
+            if (delegate.createOrWritePayload(resolved.capabilities()))
+            {
+                delegate.doFileSystemData(traceId, authorization, budgetId, reserved, flags, payload);
+            }
+            else
+            {
+                doHttpReset(traceId);
+                delegate.doFileSystemAbort(traceId, authorization);
+            }
         }
 
         private void onHttpEnd(
@@ -542,6 +570,18 @@ private void doFileSystemBegin(
                     traceId, authorization, affinity, resolved);
         }
 
+        private void doFileSystemData(
+            long traceId,
+            long authorization,
+            long budgetId,
+            int reserved,
+            int flags,
+            Flyweight payload)
+        {
+            doData(filesystem, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                traceId, authorization, budgetId, flags, reserved, payload);
+        }
+
         private void doFileSystemEnd(
             long traceId,
             long sequence,
@@ -645,13 +685,24 @@ private void onFileSystemBegin(
             {
                 final HttpBeginExFW.Builder httpBeginExBuilder =
                     httpBeginExRW.wrap(extBuffer, 0, extBuffer.capacity())
-                        .typeId(httpTypeId)
-                        .headersItem(h -> h.name(HEADER_STATUS_NAME).value(getStatus(fsBeginEx)))
+                        .typeId(httpTypeId);
+                final boolean validTag = tag != null ? tag.length() != -1 && tag.asString() != null : false;
+
+                if (createOrWritePayload(fsBeginEx.capabilities()) && validTag)
+                {
+                    httpBeginExBuilder.headersItem(h -> h.name(HEADER_STATUS_NAME).value(HEADER_STATUS_VALUE_204))
+                        .headersItem(h -> h.name(HEADER_ETAG_NAME).value(tag));
+                }
+                else
+                {
+                    httpBeginExBuilder.headersItem(h -> h.name(HEADER_STATUS_NAME).value(getStatus(fsBeginEx)))
                         .headersItem(h -> h.name(HEADER_CONTENT_TYPE_NAME).value(type))
                         .headersItem(h -> h.name(HEADER_CONTENT_LENGTH_NAME).value(length));
-                if (tag.length() != -1 && tag.asString() != null)
-                {
-                    httpBeginExBuilder.headersItem(h -> h.name(HEADER_ETAG_NAME).value(tag));
+
+                    if (validTag)
+                    {
+                        httpBeginExBuilder.headersItem(h -> h.name(HEADER_ETAG_NAME).value(tag));
+                    }
                 }
                 httpBeginEx = httpBeginExBuilder.build();
             }
@@ -675,6 +726,12 @@ private boolean canReadPayload(
             return (capabilities & READ_PAYLOAD_MASK) != 0;
         }
 
+        private boolean createOrWritePayload(
+            int capabilities)
+        {
+            return (capabilities & CREATE_PAYLOAD_MASK) != 0 || (capabilities & WRITE_PAYLOAD_MASK) != 0;
+        }
+
         private void onFileSystemData(
             DataFW data)
         {

diff --git a/...aklivity/zilla/runtime/binding/http/filesystem/internal/stream/HttpFileSystemProxyIT.java b/...aklivity/zilla/runtime/binding/http/filesystem/internal/stream/HttpFileSystemProxyIT.java
@@ -127,15 +127,6 @@ public void shouldRejectClientWithNoBinding() throws Exception
         k3po.finish();
     }
 
-    @Test
-    @Configuration("proxy.with.path.yaml")
-    @Specification({
-        "${http}/client.write.file.rejected/client"})
-    public void shouldRejectClientWriteFile() throws Exception
-    {
-        k3po.finish();
-    }
-
     @Test
     @Specification({
         "${http}/client.rejected/client"})
@@ -203,4 +194,44 @@ public void shouldReceiveClientSentAbort() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("proxy.with.path.yaml")
+    @Specification({
+        "${http}/client.create.file/client",
+        "${filesystem}/client.create.file/server"})
+    public void shouldReceiveClientCreateFile() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("proxy.with.path.yaml")
+    @Specification({
+        "${http}/client.create.existing.file.failed/client",
+        "${filesystem}/client.create.existing.file.failed/server"})
+    public void shouldRejectClientCreateExistingFileFailed() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("proxy.with.path.yaml")
+    @Specification({
+        "${http}/client.write.file/client",
+        "${filesystem}/client.write.file/server"})
+    public void shouldReceiveClientWriteFile() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("proxy.with.path.yaml")
+    @Specification({
+        "${http}/client.write.file.failed/client",
+        "${filesystem}/client.write.file.failed/server"})
+    public void shouldRejectClientWriteFileFailed() throws Exception
+    {
+        k3po.finish();
+    }
 }
diff --git a/specs/binding-filesystem.spec/src/main/resources/META-INF/zilla/filesystem.idl b/specs/binding-filesystem.spec/src/main/resources/META-INF/zilla/filesystem.idl
@@ -18,7 +18,9 @@ scope filesystem
     {
         READ_PAYLOAD,
         READ_EXTENSION,
-        READ_CHANGES
+        READ_CHANGES,
+        CREATE_PAYLOAD,
+        WRITE_PAYLOAD
     }
 
     scope stream

diff --git a/...klivity/zilla/specs/binding/filesystem/streams/application/create.file.payload/client.rpt b/...klivity/zilla/specs/binding/filesystem/streams/application/create.file.payload/client.rpt
@@ -0,0 +1,44 @@
+#
+# Copyright 2021-2024 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+connect "zilla://streams/app0"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${filesystem:beginEx()
+                                  .typeId(zilla:id("filesystem"))
+                                  .capabilities("CREATE_PAYLOAD")
+                                  .path("index_write.html")
+                                  .type("text/html")
+                                  .payloadSize(77)
+                                  .build()}
+connected
+
+write "<html>\n"
+      "<head><title>Welcome</title></head>\n"
+      "<body>Hello, world</body>\n"
+      "</html>\n"
+write flush
+
+write close
+
+read zilla:begin.ext ${filesystem:matchBeginEx()
+                                 .typeId(zilla:id("filesystem"))
+                                 .capabilities("CREATE_PAYLOAD")
+                                 .path("index_write.html")
+                                 .tag("f2c912a30f38a124c3249f8e802e0d90")
+                                 .build()}
+
+read closed