first commit with streaming support (buggy and unreliable)

gradle.properties

@@ -2,11 +2,10 @@ org.gradle.configuration-cache=false
 org.gradle.parallel=true
 org.gradle.caching=true
 
-rbcs.version = 0.1.4
+rbcs.version = 0.1.6
 
-lys.version = 2025.02.05
+lys.version = 2025.02.08
 
 gitea.maven.url = https://gitea.woggioni.net/api/packages/woggioni/maven
 docker.registry.url=gitea.woggioni.net
 
-jpms-check.configurationName = runtimeClasspath

rbcs-api/src/main/java/module-info.java

@@ -4,4 +4,5 @@ module net.woggioni.rbcs.api {
     requires io.netty.buffer;
     exports net.woggioni.rbcs.api;
     exports net.woggioni.rbcs.api.exception;
+    exports net.woggioni.rbcs.api.event;
 }

rbcs-api/src/main/java/net/woggioni/rbcs/api/Cache.java

@@ -1,14 +1,17 @@
 package net.woggioni.rbcs.api;
 
-import io.netty.buffer.ByteBuf;
-import net.woggioni.rbcs.api.exception.ContentTooLargeException;
+import io.netty.buffer.ByteBufAllocator;
 
-import java.nio.channels.ReadableByteChannel;
 import java.util.concurrent.CompletableFuture;
 
 
 public interface Cache extends AutoCloseable {
-    CompletableFuture<ReadableByteChannel> get(String key);
 
-    CompletableFuture<Void> put(String key, ByteBuf content) throws ContentTooLargeException;
+    default void get(String key, ResponseHandle responseHandle, ByteBufAllocator alloc) {
+        throw new UnsupportedOperationException();
+    }
+
+    default CompletableFuture<RequestHandle> put(String key, ResponseHandle responseHandle, ByteBufAllocator alloc) {
+        throw new UnsupportedOperationException();
+    }
 }

rbcs-api/src/main/java/net/woggioni/rbcs/api/RequestHandle.java

@@ -0,0 +1,8 @@
+package net.woggioni.rbcs.api;
+
+import net.woggioni.rbcs.api.event.RequestStreamingEvent;
+
+@FunctionalInterface
+public interface RequestHandle {
+    void handleEvent(RequestStreamingEvent evt);
+}

rbcs-api/src/main/java/net/woggioni/rbcs/api/ResponseHandle.java

@@ -0,0 +1,8 @@
+package net.woggioni.rbcs.api;
+
+import net.woggioni.rbcs.api.event.ResponseStreamingEvent;
+
+@FunctionalInterface
+public interface ResponseHandle {
+    void handleEvent(ResponseStreamingEvent evt);
+}

rbcs-api/src/main/java/net/woggioni/rbcs/api/event/RequestStreamingEvent.java

@@ -0,0 +1,26 @@
+package net.woggioni.rbcs.api.event;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+public sealed interface RequestStreamingEvent {
+
+    @Getter
+    @RequiredArgsConstructor
+    non-sealed class ChunkReceived implements RequestStreamingEvent {
+        private final ByteBuf chunk;
+    }
+
+    final class LastChunkReceived extends ChunkReceived {
+        public LastChunkReceived(ByteBuf chunk) {
+            super(chunk);
+        }
+    }
+
+    @Getter
+    @RequiredArgsConstructor
+    final class ExceptionCaught implements RequestStreamingEvent {
+        private final Throwable exception;
+    }
+}

rbcs-api/src/main/java/net/woggioni/rbcs/api/event/ResponseStreamingEvent.java

@@ -0,0 +1,42 @@
+package net.woggioni.rbcs.api.event;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.nio.channels.FileChannel;
+
+public sealed interface ResponseStreamingEvent {
+
+    final class ResponseReceived implements ResponseStreamingEvent {
+    }
+
+    @Getter
+    @RequiredArgsConstructor
+    non-sealed class ChunkReceived implements ResponseStreamingEvent {
+        private final ByteBuf chunk;
+    }
+
+    @Getter
+    @RequiredArgsConstructor
+    non-sealed class FileReceived implements ResponseStreamingEvent {
+        private final FileChannel file;
+    }
+
+    final class LastChunkReceived extends ChunkReceived {
+        public LastChunkReceived(ByteBuf chunk) {
+            super(chunk);
+        }
+    }
+
+    @Getter
+    @RequiredArgsConstructor
+    final class ExceptionCaught implements ResponseStreamingEvent {
+        private final Throwable exception;
+    }
+
+    final class NotFound implements ResponseStreamingEvent { }
+
+    NotFound NOT_FOUND = new NotFound();
+    ResponseReceived RESPONSE_RECEIVED = new ResponseReceived();
+}

rbcs-cli/build.gradle

@@ -44,7 +44,6 @@ envelopeJar {
 dependencies {
     implementation catalog.jwo
     implementation catalog.slf4j.api
-    implementation catalog.netty.codec.http
     implementation catalog.picocli
 
     implementation project(':rbcs-client')

rbcs-cli/src/main/kotlin/net/woggioni/rbcs/cli/impl/commands/BenchmarkCommand.kt

@@ -6,6 +6,8 @@ import net.woggioni.rbcs.common.contextLogger
 import net.woggioni.rbcs.common.error
 import net.woggioni.rbcs.common.info
 import net.woggioni.jwo.JWO
+import net.woggioni.jwo.LongMath
+import net.woggioni.rbcs.common.debug
 import picocli.CommandLine
 import java.security.SecureRandom
 import java.time.Duration
@@ -46,6 +48,7 @@ class BenchmarkCommand : RbcsCommand() {
             clientCommand.configuration.profiles[profileName]
                 ?: throw IllegalArgumentException("Profile $profileName does not exist in configuration")
         }
+        val progressThreshold = LongMath.ceilDiv(numberOfEntries.toLong(), 20)
         RemoteBuildCacheClient(profile).use { client ->
 
             val entryGenerator = sequence {
@@ -79,7 +82,12 @@ class BenchmarkCommand : RbcsCommand() {
                                 completionQueue.put(result)
                             }
                             semaphore.release()
-                            completionCounter.incrementAndGet()
+                            val completed = completionCounter.incrementAndGet()
+                            if(completed.mod(progressThreshold) == 0L) {
+                                log.debug {
+                                    "Inserted $completed / $numberOfEntries"
+                                }
+                            }
                         }
                     } else {
                         Thread.sleep(0)
@@ -121,7 +129,12 @@ class BenchmarkCommand : RbcsCommand() {
                             }
                         }
                         future.whenComplete { _, _ ->
-                            completionCounter.incrementAndGet()
+                            val completed = completionCounter.incrementAndGet()
+                            if(completed.mod(progressThreshold) == 0L) {
+                                log.debug {
+                                    "Retrieved $completed / ${entries.size}"
+                                }
+                            }
                             semaphore.release()
                         }
                     } else {

rbcs-client/build.gradle

@@ -6,9 +6,11 @@ plugins {
 dependencies {
     implementation project(':rbcs-api')
     implementation project(':rbcs-common')
-    implementation catalog.picocli
     implementation catalog.slf4j.api
     implementation catalog.netty.buffer
+    implementation catalog.netty.handler
+    implementation catalog.netty.transport
+    implementation catalog.netty.common
     implementation catalog.netty.codec.http
 
     testRuntimeOnly catalog.logback.classic

rbcs-client/src/main/kotlin/net/woggioni/rbcs/client/Client.kt

@@ -45,9 +45,9 @@ import java.time.Duration
 import java.util.Base64
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.atomic.AtomicInteger
+import kotlin.random.Random
 import io.netty.util.concurrent.Future as NettyFuture
 
-
 class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoCloseable {
     private val group: NioEventLoopGroup
     private var sslContext: SslContext
@@ -206,6 +206,7 @@ class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoC
                 retryPolicy.initialDelayMillis.toDouble(),
                 retryPolicy.exp,
                 outcomeHandler,
+                Random.Default,
                 operation
             )
         } else {

rbcs-client/src/main/kotlin/net/woggioni/rbcs/client/retry.kt

@@ -3,6 +3,8 @@ package net.woggioni.rbcs.client
 import io.netty.util.concurrent.EventExecutorGroup
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.TimeUnit
+import kotlin.math.pow
+import kotlin.random.Random
 
 sealed class OperationOutcome<T> {
     class Success<T>(val result: T) : OperationOutcome<T>()
@@ -24,8 +26,10 @@ fun <T> executeWithRetry(
     initialDelay: Double,
     exp: Double,
     outcomeHandler: OutcomeHandler<T>,
+    randomizer : Random?,
     cb: () -> CompletableFuture<T>
 ): CompletableFuture<T> {
+
     val finalResult = cb()
     var future = finalResult
     var shortCircuit = false
@@ -46,7 +50,7 @@ fun <T> executeWithRetry(
                     is OutcomeHandlerResult.Retry -> {
                         val res = CompletableFuture<T>()
                         val delay = run {
-                            val scheduledDelay = (initialDelay * Math.pow(exp, i.toDouble())).toLong()
+                            val scheduledDelay = (initialDelay * exp.pow(i.toDouble()) * (1.0 + (randomizer?.nextDouble(-0.5, 0.5) ?: 0.0))).toLong()
                             outcomeHandlerResult.suggestedDelayMillis?.coerceAtMost(scheduledDelay) ?: scheduledDelay
                         }
                         eventExecutorGroup.schedule({

rbcs-client/src/test/kotlin/net/woggioni/rbcs/client/RetryTest.kt

@@ -89,7 +89,7 @@ class RetryTest {
         val random = Random(testArgs.seed)
 
         val future =
-            executeWithRetry(executor, testArgs.maxAttempt, testArgs.initialDelay, testArgs.exp, outcomeHandler) {
+            executeWithRetry(executor, testArgs.maxAttempt, testArgs.initialDelay, testArgs.exp, outcomeHandler, null) {
                 val now = System.nanoTime()
                 val result = CompletableFuture<Int>()
                 executor.submit {
@@ -129,7 +129,7 @@ class RetryTest {
                     previousAttempt.first + testArgs.initialDelay * Math.pow(testArgs.exp, index.toDouble()) * 1e6
                 val actualTimestamp = timestamp
                 val err = Math.abs(expectedTimestamp - actualTimestamp) / expectedTimestamp
-                Assertions.assertTrue(err < 1e-3)
+                Assertions.assertTrue(err < 1e-2)
             }
             if (index == attempts.size - 1 && index < testArgs.maxAttempt - 1) {
                 /*

rbcs-common/src/main/kotlin/net/woggioni/rbcs/common/BB.kt

@@ -0,0 +1,15 @@
+package net.woggioni.rbcs.common
+
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
+import io.netty.buffer.CompositeByteBuf
+
+fun extractChunk(buf: CompositeByteBuf, alloc: ByteBufAllocator): ByteBuf {
+    val chunk = alloc.compositeBuffer()
+    for (component in buf.decompose(0, buf.readableBytes())) {
+        chunk.addComponent(true, component.retain())
+    }
+    buf.removeComponents(0, buf.numComponents())
+    buf.clear()
+    return chunk
+}

rbcs-common/src/main/kotlin/net/woggioni/rbcs/common/RBCS.kt

@@ -12,6 +12,24 @@ object RBCS {
     const val RBCS_PREFIX: String = "rbcs"
     const val XML_SCHEMA_NAMESPACE_URI = "http://www.w3.org/2001/XMLSchema-instance"
 
+    fun ByteArray.toInt(index : Int = 0) : Long {
+        if(index + 4 > size) throw IllegalArgumentException("Not enough bytes to decode a 32 bits integer")
+        var value : Long = 0
+        for (b in index until index + 4) {
+            value = (value shl 8) + (get(b).toInt() and 0xFF)
+        }
+        return value
+    }
+
+    fun ByteArray.toLong(index : Int = 0) : Long {
+        if(index + 8 > size) throw IllegalArgumentException("Not enough bytes to decode a 64 bits long integer")
+        var value : Long = 0
+        for (b in index until index + 8) {
+            value = (value shl 8) + (get(b).toInt() and 0xFF)
+        }
+        return value
+    }
+
     fun digest(
         data: ByteArray,
         md: MessageDigest = MessageDigest.getInstance("MD5")

rbcs-server-memcache/build.gradle

@@ -34,6 +34,7 @@ dependencies {
     implementation catalog.jwo
     implementation catalog.slf4j.api
     implementation catalog.netty.common
+    implementation catalog.netty.handler
     implementation catalog.netty.codec.memcache
 
     bundle catalog.netty.codec.memcache

rbcs-server-memcache/src/main/java/module-info.java

@@ -11,6 +11,7 @@ module net.woggioni.rbcs.server.memcache {
     requires io.netty.codec.memcache;
     requires io.netty.common;
     requires io.netty.buffer;
+    requires io.netty.handler;
     requires org.slf4j;
 
     provides CacheProvider with net.woggioni.rbcs.server.memcache.MemcacheCacheProvider;

rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/MemcacheCache.kt

@@ -1,20 +1,232 @@
 package net.woggioni.rbcs.server.memcache
 
-import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
+import io.netty.buffer.Unpooled
+import io.netty.handler.codec.memcache.binary.BinaryMemcacheOpcodes
+import io.netty.handler.codec.memcache.binary.BinaryMemcacheResponseStatus
+import io.netty.handler.codec.memcache.binary.DefaultBinaryMemcacheRequest
 import net.woggioni.rbcs.api.Cache
+import net.woggioni.rbcs.api.RequestHandle
+import net.woggioni.rbcs.api.ResponseHandle
+import net.woggioni.rbcs.api.event.RequestStreamingEvent
+import net.woggioni.rbcs.api.event.ResponseStreamingEvent
+import net.woggioni.rbcs.api.exception.ContentTooLargeException
+import net.woggioni.rbcs.common.ByteBufOutputStream
+import net.woggioni.rbcs.common.RBCS.digest
+import net.woggioni.rbcs.common.contextLogger
+import net.woggioni.rbcs.common.debug
+import net.woggioni.rbcs.common.extractChunk
 import net.woggioni.rbcs.server.memcache.client.MemcacheClient
-import java.nio.channels.ReadableByteChannel
+import net.woggioni.rbcs.server.memcache.client.MemcacheResponseHandle
+import net.woggioni.rbcs.server.memcache.client.StreamingRequestEvent
+import net.woggioni.rbcs.server.memcache.client.StreamingResponseEvent
+import java.security.MessageDigest
+import java.time.Duration
+import java.time.Instant
 import java.util.concurrent.CompletableFuture
+import java.util.zip.Deflater
+import java.util.zip.DeflaterOutputStream
+import java.util.zip.Inflater
+import java.util.zip.InflaterOutputStream
 
-class MemcacheCache(private val cfg : MemcacheCacheConfiguration) : Cache {
-    private val memcacheClient = MemcacheClient(cfg)
+class MemcacheCache(private val cfg: MemcacheCacheConfiguration) : Cache {
 
-    override fun get(key: String): CompletableFuture<ReadableByteChannel?> {
-        return memcacheClient.get(key)
+    companion object {
+        @JvmStatic
+        private val log = contextLogger()
     }
 
-    override fun put(key: String, content: ByteBuf): CompletableFuture<Void> {
-        return memcacheClient.put(key, content, cfg.maxAge)
+    private val memcacheClient = MemcacheClient(cfg)
+
+    override fun get(key: String, responseHandle: ResponseHandle, alloc: ByteBufAllocator) {
+        val compressionMode = cfg.compressionMode
+        val buf = alloc.compositeBuffer()
+        val stream = ByteBufOutputStream(buf).let { outputStream ->
+            if (compressionMode != null) {
+                when (compressionMode) {
+                    MemcacheCacheConfiguration.CompressionMode.DEFLATE -> {
+                        InflaterOutputStream(
+                            outputStream,
+                            Inflater()
+                        )
+                    }
+                }
+            } else {
+                outputStream
+            }
+        }
+        val memcacheResponseHandle = object : MemcacheResponseHandle {
+            override fun handleEvent(evt: StreamingResponseEvent) {
+                when (evt) {
+                    is StreamingResponseEvent.ResponseReceived -> {
+                        if (evt.response.status() == BinaryMemcacheResponseStatus.SUCCESS) {
+                            responseHandle.handleEvent(ResponseStreamingEvent.RESPONSE_RECEIVED)
+                        } else if (evt.response.status() == BinaryMemcacheResponseStatus.KEY_ENOENT) {
+                            responseHandle.handleEvent(ResponseStreamingEvent.NOT_FOUND)
+                        } else {
+                            responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(MemcacheException(evt.response.status())))
+                        }
+                    }
+
+                    is StreamingResponseEvent.LastContentReceived -> {
+                        evt.content.content().let { content ->
+                            content.readBytes(stream, content.readableBytes())
+                        }
+                        buf.retain()
+                        stream.close()
+                        val chunk = extractChunk(buf, alloc)
+                        buf.release()
+                        responseHandle.handleEvent(
+                            ResponseStreamingEvent.LastChunkReceived(
+                                chunk
+                            )
+                        )
+                    }
+
+                    is StreamingResponseEvent.ContentReceived -> {
+                        evt.content.content().let { content ->
+                            content.readBytes(stream, content.readableBytes())
+                        }
+                        if (buf.readableBytes() >= cfg.chunkSize) {
+                            val chunk = extractChunk(buf, alloc)
+                            responseHandle.handleEvent(ResponseStreamingEvent.ChunkReceived(chunk))
+                        }
+                    }
+
+                    is StreamingResponseEvent.ExceptionCaught -> {
+                        responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(evt.exception))
+                    }
+                }
+            }
+        }
+        memcacheClient.sendRequest(Unpooled.wrappedBuffer(key.toByteArray()), memcacheResponseHandle)
+            .thenApply { memcacheRequestHandle ->
+                val request = (cfg.digestAlgorithm
+                    ?.let(MessageDigest::getInstance)
+                    ?.let { md ->
+                        digest(key.toByteArray(), md)
+                    } ?: key.toByteArray(Charsets.UTF_8)
+                        ).let { digest ->
+                        DefaultBinaryMemcacheRequest(Unpooled.wrappedBuffer(digest)).apply {
+                            setOpcode(BinaryMemcacheOpcodes.GET)
+                        }
+                    }
+                memcacheRequestHandle.handleEvent(StreamingRequestEvent.SendRequest(request))
+            }.exceptionally { ex ->
+                responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(ex))
+            }
+    }
+
+    private fun encodeExpiry(expiry: Duration): Int {
+        val expirySeconds = expiry.toSeconds()
+        return expirySeconds.toInt().takeIf { it.toLong() == expirySeconds }
+            ?: Instant.ofEpochSecond(expirySeconds).epochSecond.toInt()
+    }
+
+    override fun put(
+        key: String,
+        responseHandle: ResponseHandle,
+        alloc: ByteBufAllocator
+    ): CompletableFuture<RequestHandle> {
+        val memcacheResponseHandle = object : MemcacheResponseHandle {
+            override fun handleEvent(evt: StreamingResponseEvent) {
+                when (evt) {
+                    is StreamingResponseEvent.ResponseReceived -> {
+                        when (evt.response.status()) {
+                            BinaryMemcacheResponseStatus.SUCCESS -> {
+                                responseHandle.handleEvent(ResponseStreamingEvent.RESPONSE_RECEIVED)
+                            }
+
+                            BinaryMemcacheResponseStatus.KEY_ENOENT -> {
+                                responseHandle.handleEvent(ResponseStreamingEvent.NOT_FOUND)
+                            }
+
+                            BinaryMemcacheResponseStatus.E2BIG -> {
+                                responseHandle.handleEvent(
+                                    ResponseStreamingEvent.ExceptionCaught(
+                                        ContentTooLargeException("Request payload is too big", null)
+                                    )
+                                )
+                            }
+
+                            else -> {
+                                responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(MemcacheException(evt.response.status())))
+                            }
+                        }
+                    }
+
+                    is StreamingResponseEvent.LastContentReceived -> {
+                        responseHandle.handleEvent(
+                            ResponseStreamingEvent.LastChunkReceived(
+                                evt.content.content().retain()
+                            )
+                        )
+                    }
+
+                    is StreamingResponseEvent.ContentReceived -> {
+                        responseHandle.handleEvent(ResponseStreamingEvent.ChunkReceived(evt.content.content().retain()))
+                    }
+
+                    is StreamingResponseEvent.ExceptionCaught -> {
+                        responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(evt.exception))
+                    }
+                }
+            }
+        }
+        val result: CompletableFuture<RequestHandle> =
+            memcacheClient.sendRequest(Unpooled.wrappedBuffer(key.toByteArray()), memcacheResponseHandle)
+                .thenApply { memcacheRequestHandle ->
+                    val request = (cfg.digestAlgorithm
+                        ?.let(MessageDigest::getInstance)
+                        ?.let { md ->
+                            digest(key.toByteArray(), md)
+                        } ?: key.toByteArray(Charsets.UTF_8)).let { digest ->
+                        val extras = Unpooled.buffer(8, 8)
+                        extras.writeInt(0)
+                        extras.writeInt(encodeExpiry(cfg.maxAge))
+                        DefaultBinaryMemcacheRequest(Unpooled.wrappedBuffer(digest), extras).apply {
+                            setOpcode(BinaryMemcacheOpcodes.SET)
+                        }
+                    }
+//                    memcacheRequestHandle.handleEvent(StreamingRequestEvent.SendRequest(request))
+                    val compressionMode = cfg.compressionMode
+                    val buf = alloc.heapBuffer()
+                    val stream = ByteBufOutputStream(buf).let { outputStream ->
+                        if (compressionMode != null) {
+                            when (compressionMode) {
+                                MemcacheCacheConfiguration.CompressionMode.DEFLATE -> {
+                                    DeflaterOutputStream(
+                                        outputStream,
+                                        Deflater(Deflater.DEFAULT_COMPRESSION, false)
+                                    )
+                                }
+                            }
+                        } else {
+                            outputStream
+                        }
+                    }
+                    RequestHandle { evt ->
+                        when (evt) {
+                            is RequestStreamingEvent.LastChunkReceived -> {
+                                evt.chunk.readBytes(stream, evt.chunk.readableBytes())
+                                buf.retain()
+                                stream.close()
+                                request.setTotalBodyLength(buf.readableBytes() + request.keyLength() + request.extrasLength())
+                                memcacheRequestHandle.handleEvent(StreamingRequestEvent.SendRequest(request))
+                                memcacheRequestHandle.handleEvent(StreamingRequestEvent.SendLastChunk(buf))
+                            }
+
+                            is RequestStreamingEvent.ChunkReceived -> {
+                                evt.chunk.readBytes(stream, evt.chunk.readableBytes())
+                            }
+
+                            is RequestStreamingEvent.ExceptionCaught -> {
+                                stream.close()
+                            }
+                        }
+                    }
+                }
+        return result
     }
 
     override fun close() {

rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/MemcacheCacheConfiguration.kt

@@ -10,14 +10,10 @@ data class MemcacheCacheConfiguration(
     val maxSize: Int = 0x100000,
     val digestAlgorithm: String? = null,
     val compressionMode: CompressionMode? = null,
+    val chunkSize : Int
 ) : Configuration.Cache {
 
     enum class CompressionMode {
-        /**
-         * Gzip mode
-         */
-        GZIP,
-
         /**
          * Deflate mode
          */

rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/MemcacheCacheProvider.kt

@@ -29,12 +29,14 @@ class MemcacheCacheProvider : CacheProvider<MemcacheCacheConfiguration> {
             ?.let(Duration::parse)
             ?: Duration.ofDays(1)
         val maxSize = el.renderAttribute("max-size")
-            ?.let(String::toInt)
+            ?.let(Integer::decode)
             ?: 0x100000
+        val chunkSize = el.renderAttribute("chunk-size")
+            ?.let(Integer::decode)
+            ?: 0x4000
         val compressionMode = el.renderAttribute("compression-mode")
             ?.let {
                 when (it) {
-                    "gzip" -> MemcacheCacheConfiguration.CompressionMode.GZIP
                     "deflate" -> MemcacheCacheConfiguration.CompressionMode.DEFLATE
                     else -> MemcacheCacheConfiguration.CompressionMode.DEFLATE
                 }
@@ -63,6 +65,7 @@ class MemcacheCacheProvider : CacheProvider<MemcacheCacheConfiguration> {
             maxSize,
             digestAlgorithm,
             compressionMode,
+            chunkSize
         )
     }
 
@@ -70,7 +73,6 @@ class MemcacheCacheProvider : CacheProvider<MemcacheCacheConfiguration> {
         val result = doc.createElement("cache")
         Xml.of(doc, result) {
             attr("xmlns:${xmlNamespacePrefix}", xmlNamespace, namespaceURI = "http://www.w3.org/2000/xmlns/")
-
             attr("xs:type", "${xmlNamespacePrefix}:$xmlType", RBCS.XML_SCHEMA_NAMESPACE_URI)
             for (server in servers) {
                 node("server") {
@@ -84,13 +86,13 @@ class MemcacheCacheProvider : CacheProvider<MemcacheCacheConfiguration> {
             }
             attr("max-age", maxAge.toString())
             attr("max-size", maxSize.toString())
+            attr("chunk-size", chunkSize.toString())
             digestAlgorithm?.let { digestAlgorithm ->
                 attr("digest", digestAlgorithm)
             }
             compressionMode?.let { compressionMode ->
                 attr(
                     "compression-mode", when (compressionMode) {
-                        MemcacheCacheConfiguration.CompressionMode.GZIP -> "gzip"
                         MemcacheCacheConfiguration.CompressionMode.DEFLATE -> "deflate"
                     }
                 )

rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/client/Event.kt

@@ -0,0 +1,30 @@
+package net.woggioni.rbcs.server.memcache.client
+
+import io.netty.buffer.ByteBuf
+import io.netty.handler.codec.memcache.LastMemcacheContent
+import io.netty.handler.codec.memcache.MemcacheContent
+import io.netty.handler.codec.memcache.binary.BinaryMemcacheRequest
+import io.netty.handler.codec.memcache.binary.BinaryMemcacheResponse
+
+sealed interface StreamingRequestEvent {
+    class SendRequest(val request : BinaryMemcacheRequest) : StreamingRequestEvent
+    open class SendChunk(val chunk : ByteBuf) : StreamingRequestEvent
+    class SendLastChunk(chunk : ByteBuf) : SendChunk(chunk)
+    class ExceptionCaught(val exception : Throwable) : StreamingRequestEvent
+}
+
+sealed interface StreamingResponseEvent {
+    class ResponseReceived(val response : BinaryMemcacheResponse) : StreamingResponseEvent
+    open class ContentReceived(val content : MemcacheContent) : StreamingResponseEvent
+    class LastContentReceived(val lastContent : LastMemcacheContent) : ContentReceived(lastContent)
+    class ExceptionCaught(val exception : Throwable) : StreamingResponseEvent
+}
+
+interface MemcacheRequestHandle {
+    fun handleEvent(evt : StreamingRequestEvent)
+}
+
+interface MemcacheResponseHandle {
+    fun handleEvent(evt : StreamingResponseEvent)
+}
+

rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/client/MemcacheClient.kt

@@ -3,7 +3,6 @@ package net.woggioni.rbcs.server.memcache.client
 
 import io.netty.bootstrap.Bootstrap
 import io.netty.buffer.ByteBuf
-import io.netty.buffer.Unpooled
 import io.netty.channel.Channel
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelOption
@@ -14,36 +13,23 @@ import io.netty.channel.pool.AbstractChannelPoolHandler
 import io.netty.channel.pool.ChannelPool
 import io.netty.channel.pool.FixedChannelPool
 import io.netty.channel.socket.nio.NioSocketChannel
-import io.netty.handler.codec.DecoderException
+import io.netty.handler.codec.memcache.DefaultLastMemcacheContent
+import io.netty.handler.codec.memcache.DefaultMemcacheContent
+import io.netty.handler.codec.memcache.LastMemcacheContent
+import io.netty.handler.codec.memcache.MemcacheContent
+import io.netty.handler.codec.memcache.MemcacheObject
 import io.netty.handler.codec.memcache.binary.BinaryMemcacheClientCodec
-import io.netty.handler.codec.memcache.binary.BinaryMemcacheObjectAggregator
-import io.netty.handler.codec.memcache.binary.BinaryMemcacheOpcodes
-import io.netty.handler.codec.memcache.binary.BinaryMemcacheResponseStatus
-import io.netty.handler.codec.memcache.binary.DefaultFullBinaryMemcacheRequest
-import io.netty.handler.codec.memcache.binary.FullBinaryMemcacheRequest
-import io.netty.handler.codec.memcache.binary.FullBinaryMemcacheResponse
+import io.netty.handler.codec.memcache.binary.BinaryMemcacheResponse
+import io.netty.handler.logging.LoggingHandler
 import io.netty.util.concurrent.GenericFutureListener
-import net.woggioni.rbcs.common.ByteBufInputStream
-import net.woggioni.rbcs.common.ByteBufOutputStream
-import net.woggioni.rbcs.common.RBCS.digest
 import net.woggioni.rbcs.common.HostAndPort
 import net.woggioni.rbcs.common.contextLogger
+import net.woggioni.rbcs.common.debug
 import net.woggioni.rbcs.server.memcache.MemcacheCacheConfiguration
-import net.woggioni.rbcs.server.memcache.MemcacheException
-import net.woggioni.jwo.JWO
 import java.net.InetSocketAddress
-import java.nio.channels.Channels
-import java.nio.channels.ReadableByteChannel
-import java.security.MessageDigest
-import java.time.Duration
-import java.time.Instant
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ConcurrentHashMap
-import java.util.zip.Deflater
-import java.util.zip.DeflaterOutputStream
-import java.util.zip.GZIPInputStream
-import java.util.zip.GZIPOutputStream
-import java.util.zip.InflaterInputStream
+import java.util.concurrent.atomic.AtomicLong
 import io.netty.util.concurrent.Future as NettyFuture
 
 
@@ -61,6 +47,8 @@ class MemcacheClient(private val cfg: MemcacheCacheConfiguration) : AutoCloseabl
         group = NioEventLoopGroup()
     }
 
+    private val counter = AtomicLong(0)
+
     private fun newConnectionPool(server: MemcacheCacheConfiguration.Server): FixedChannelPool {
         val bootstrap = Bootstrap().apply {
             group(group)
@@ -76,18 +64,15 @@ class MemcacheClient(private val cfg: MemcacheCacheConfiguration) : AutoCloseabl
             override fun channelCreated(ch: Channel) {
                 val pipeline: ChannelPipeline = ch.pipeline()
                 pipeline.addLast(BinaryMemcacheClientCodec())
-                pipeline.addLast(BinaryMemcacheObjectAggregator(Integer.MAX_VALUE))
+                pipeline.addLast(LoggingHandler())
             }
         }
         return FixedChannelPool(bootstrap, channelPoolHandler, server.maxConnections)
     }
 
-
-    private fun sendRequest(request: FullBinaryMemcacheRequest): CompletableFuture<FullBinaryMemcacheResponse> {
-
+    fun sendRequest(key: ByteBuf, responseHandle: MemcacheResponseHandle): CompletableFuture<MemcacheRequestHandle> {
         val server = cfg.servers.let { servers ->
             if (servers.size > 1) {
-                val key = request.key().duplicate()
                 var checksum = 0
                 while (key.readableBytes() > 4) {
                     val byte = key.readInt()
@@ -103,7 +88,7 @@ class MemcacheClient(private val cfg: MemcacheCacheConfiguration) : AutoCloseabl
             }
         }
 
-        val response = CompletableFuture<FullBinaryMemcacheResponse>()
+        val response = CompletableFuture<MemcacheRequestHandle>()
         // Custom handler for processing responses
         val pool = connectionPool.computeIfAbsent(server.endpoint) {
             newConnectionPool(server)
@@ -113,31 +98,73 @@ class MemcacheClient(private val cfg: MemcacheCacheConfiguration) : AutoCloseabl
                 if (channelFuture.isSuccess) {
                     val channel = channelFuture.now
                     val pipeline = channel.pipeline()
-                    channel.pipeline()
-                        .addLast("client-handler", object : SimpleChannelInboundHandler<FullBinaryMemcacheResponse>() {
+                    val handler = object : SimpleChannelInboundHandler<MemcacheObject>() {
                         override fun channelRead0(
                             ctx: ChannelHandlerContext,
-                                msg: FullBinaryMemcacheResponse
+                            msg: MemcacheObject
                         ) {
+                            when (msg) {
+                                is BinaryMemcacheResponse -> responseHandle.handleEvent(
+                                    StreamingResponseEvent.ResponseReceived(
+                                        msg
+                                    )
+                                )
+
+                                is LastMemcacheContent -> {
+                                    responseHandle.handleEvent(
+                                        StreamingResponseEvent.LastContentReceived(
+                                            msg
+                                        )
+                                    )
                                     pipeline.removeLast()
                                     pool.release(channel)
-                                msg.touch("The method's caller must remember to release this")
-                                response.complete(msg.retain())
+                                }
+
+                                is MemcacheContent -> responseHandle.handleEvent(
+                                    StreamingResponseEvent.ContentReceived(
+                                        msg
+                                    )
+                                )
+                            }
                         }
 
                         override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
-                                val ex = when (cause) {
-                                    is DecoderException -> cause.cause!!
-                                    else -> cause
-                                }
+                            responseHandle.handleEvent(StreamingResponseEvent.ExceptionCaught(cause))
                             ctx.close()
                             pipeline.removeLast()
                             pool.release(channel)
-                                response.completeExceptionally(ex)
+                        }
+                    }
+                    channel.pipeline()
+                        .addLast("client-handler", handler)
+                    response.complete(object : MemcacheRequestHandle {
+                        override fun handleEvent(evt: StreamingRequestEvent) {
+                            when (evt) {
+                                is StreamingRequestEvent.SendRequest -> {
+                                    channel.writeAndFlush(evt.request)
+                                }
+
+                                is StreamingRequestEvent.SendLastChunk -> {
+                                    channel.writeAndFlush(DefaultLastMemcacheContent(evt.chunk))
+                                    val value = counter.incrementAndGet()
+                                    log.debug {
+                                        "Finished request counter: $value"
+                                    }
+                                }
+
+                                is StreamingRequestEvent.SendChunk -> {
+                                    channel.writeAndFlush(DefaultMemcacheContent(evt.chunk))
+                                }
+
+                                is StreamingRequestEvent.ExceptionCaught -> {
+                                    responseHandle.handleEvent(StreamingResponseEvent.ExceptionCaught(evt.exception))
+                                    channel.close()
+                                    pipeline.removeLast()
+                                    pool.release(channel)
+                                }
+                            }
                         }
                     })
-                    request.touch()
-                    channel.writeAndFlush(request)
                 } else {
                     response.completeExceptionally(channelFuture.cause())
                 }
@@ -146,107 +173,6 @@ class MemcacheClient(private val cfg: MemcacheCacheConfiguration) : AutoCloseabl
         return response
     }
 
-    private fun encodeExpiry(expiry: Duration): Int {
-        val expirySeconds = expiry.toSeconds()
-        return expirySeconds.toInt().takeIf { it.toLong() == expirySeconds }
-            ?: Instant.ofEpochSecond(expirySeconds).epochSecond.toInt()
-    }
-
-    fun get(key: String): CompletableFuture<ReadableByteChannel?> {
-        val request = (cfg.digestAlgorithm
-            ?.let(MessageDigest::getInstance)
-            ?.let { md ->
-                digest(key.toByteArray(), md)
-            } ?: key.toByteArray(Charsets.UTF_8)).let { digest ->
-            DefaultFullBinaryMemcacheRequest(Unpooled.wrappedBuffer(digest), null).apply {
-                setOpcode(BinaryMemcacheOpcodes.GET)
-            }
-        }
-        return sendRequest(request).thenApply { response ->
-            try {
-                when (val status = response.status()) {
-                    BinaryMemcacheResponseStatus.SUCCESS -> {
-                        val compressionMode = cfg.compressionMode
-                        val content = response.content().retain()
-                        content.touch()
-                        if (compressionMode != null) {
-                            when (compressionMode) {
-                                MemcacheCacheConfiguration.CompressionMode.GZIP -> {
-                                    GZIPInputStream(ByteBufInputStream(content))
-                                }
-
-                                MemcacheCacheConfiguration.CompressionMode.DEFLATE -> {
-                                    InflaterInputStream(ByteBufInputStream(content))
-                                }
-                            }
-                        } else {
-                            ByteBufInputStream(content)
-                        }.let(Channels::newChannel)
-                    }
-
-                    BinaryMemcacheResponseStatus.KEY_ENOENT -> {
-                        null
-                    }
-
-                    else -> throw MemcacheException(status)
-                }
-            } finally {
-                response.release()
-            }
-        }
-    }
-
-    fun put(key: String, content: ByteBuf, expiry: Duration, cas: Long? = null): CompletableFuture<Void> {
-        val request = (cfg.digestAlgorithm
-            ?.let(MessageDigest::getInstance)
-            ?.let { md ->
-                digest(key.toByteArray(), md)
-            } ?: key.toByteArray(Charsets.UTF_8)).let { digest ->
-            val extras = Unpooled.buffer(8, 8)
-            extras.writeInt(0)
-            extras.writeInt(encodeExpiry(expiry))
-            val compressionMode = cfg.compressionMode
-            content.retain()
-            val payload = if (compressionMode != null) {
-                val inputStream = ByteBufInputStream(content)
-                val buf = content.alloc().buffer()
-                buf.retain()
-                val outputStream = when (compressionMode) {
-                    MemcacheCacheConfiguration.CompressionMode.GZIP -> {
-                        GZIPOutputStream(ByteBufOutputStream(buf))
-                    }
-
-                    MemcacheCacheConfiguration.CompressionMode.DEFLATE -> {
-                        DeflaterOutputStream(ByteBufOutputStream(buf), Deflater(Deflater.DEFAULT_COMPRESSION, false))
-                    }
-                }
-                inputStream.use { i ->
-                    outputStream.use { o ->
-                        JWO.copy(i, o)
-                    }
-                }
-                buf
-            } else {
-                content
-            }
-            DefaultFullBinaryMemcacheRequest(Unpooled.wrappedBuffer(digest), extras, payload).apply {
-                setOpcode(BinaryMemcacheOpcodes.SET)
-                cas?.let(this::setCas)
-            }
-        }
-        return sendRequest(request).thenApply { response ->
-            try {
-                when (val status = response.status()) {
-                    BinaryMemcacheResponseStatus.SUCCESS -> null
-                    else -> throw MemcacheException(status)
-                }
-            } finally {
-                response.release()
-            }
-        }
-    }
-
-
     fun shutDown(): NettyFuture<*> {
         return group.shutdownGracefully()
     }

rbcs-server-memcache/src/main/resources/net/woggioni/rbcs/server/memcache/schema/rbcs-memcache.xsd

@@ -20,7 +20,8 @@
                     <xs:element name="server" type="rbcs-memcache:memcacheServerType"/>
                 </xs:sequence>
                 <xs:attribute name="max-age" type="xs:duration" default="P1D"/>
-                <xs:attribute name="max-size" type="xs:unsignedInt" default="1048576"/>
+                <xs:attribute name="max-size" type="rbcs:byteSize" default="1048576"/>
+                <xs:attribute name="chunk-size" type="rbcs:byteSize" default="0x4000"/>
                 <xs:attribute name="digest" type="xs:token" />
                 <xs:attribute name="compression-mode" type="rbcs-memcache:compressionType"/>
             </xs:extension>
@@ -30,7 +31,6 @@
     <xs:simpleType name="compressionType">
         <xs:restriction base="xs:token">
             <xs:enumeration value="deflate"/>
-            <xs:enumeration value="gzip"/>
         </xs:restriction>
     </xs:simpleType>
 

rbcs-server/build.gradle

@@ -9,6 +9,9 @@ dependencies {
     implementation catalog.jwo
     implementation catalog.slf4j.api
     implementation catalog.netty.codec.http
+    implementation catalog.netty.handler
+    implementation catalog.netty.buffer
+    implementation catalog.netty.transport
 
     api project(':rbcs-common')
     api project(':rbcs-api')
@@ -36,3 +39,4 @@ publishing {
 }
 
 
+

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/RemoteBuildCacheServer.kt

@@ -16,7 +16,6 @@ import io.netty.handler.codec.compression.CompressionOptions
 import io.netty.handler.codec.http.DefaultHttpContent
 import io.netty.handler.codec.http.HttpContentCompressor
 import io.netty.handler.codec.http.HttpHeaderNames
-import io.netty.handler.codec.http.HttpObjectAggregator
 import io.netty.handler.codec.http.HttpRequest
 import io.netty.handler.codec.http.HttpServerCodec
 import io.netty.handler.ssl.ClientAuth
@@ -30,11 +29,13 @@ import io.netty.handler.timeout.IdleStateHandler
 import io.netty.util.AttributeKey
 import io.netty.util.concurrent.DefaultEventExecutorGroup
 import io.netty.util.concurrent.EventExecutorGroup
+import net.woggioni.jwo.JWO
+import net.woggioni.jwo.Tuple2
 import net.woggioni.rbcs.api.Configuration
 import net.woggioni.rbcs.api.exception.ConfigurationException
-import net.woggioni.rbcs.common.RBCS.toUrl
 import net.woggioni.rbcs.common.PasswordSecurity.decodePasswordHash
 import net.woggioni.rbcs.common.PasswordSecurity.hashPassword
+import net.woggioni.rbcs.common.RBCS.toUrl
 import net.woggioni.rbcs.common.Xml
 import net.woggioni.rbcs.common.contextLogger
 import net.woggioni.rbcs.common.debug
@@ -48,8 +49,6 @@ import net.woggioni.rbcs.server.configuration.Serializer
 import net.woggioni.rbcs.server.exception.ExceptionHandler
 import net.woggioni.rbcs.server.handler.ServerHandler
 import net.woggioni.rbcs.server.throttling.ThrottlingHandler
-import net.woggioni.jwo.JWO
-import net.woggioni.jwo.Tuple2
 import java.io.OutputStream
 import java.net.InetSocketAddress
 import java.nio.file.Files
@@ -59,6 +58,7 @@ import java.security.PrivateKey
 import java.security.cert.X509Certificate
 import java.util.Arrays
 import java.util.Base64
+import java.util.concurrent.Future
 import java.util.concurrent.TimeUnit
 import java.util.regex.Matcher
 import java.util.regex.Pattern
@@ -128,11 +128,12 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
                     val clientCertificate = peerCertificates.first() as X509Certificate
                     val user = userExtractor?.extract(clientCertificate)
                     val group = groupExtractor?.extract(clientCertificate)
-                    val allGroups = ((user?.groups ?: emptySet()).asSequence() + sequenceOf(group).filterNotNull()).toSet()
+                    val allGroups =
+                        ((user?.groups ?: emptySet()).asSequence() + sequenceOf(group).filterNotNull()).toSet()
                     AuthenticationResult(user, allGroups)
-                } ?: anonymousUserGroups?.let{ AuthenticationResult(null, it) }
+                } ?: anonymousUserGroups?.let { AuthenticationResult(null, it) }
             } catch (es: SSLPeerUnverifiedException) {
-                anonymousUserGroups?.let{ AuthenticationResult(null, it) }
+                anonymousUserGroups?.let { AuthenticationResult(null, it) }
             }
         }
     }
@@ -191,7 +192,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
     private class ServerInitializer(
         private val cfg: Configuration,
         private val eventExecutorGroup: EventExecutorGroup
-    ) : ChannelInitializer<Channel>() {
+    ) : ChannelInitializer<Channel>(), AutoCloseable {
 
         companion object {
             private fun createSslCtx(tls: Configuration.Tls): SslContext {
@@ -213,7 +214,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
                             trustManager(
                                 ClientCertificateValidator.getTrustManager(ts, trustStore.isCheckCertificateStatus)
                             )
-                            if(trustStore.isRequireClientCertificate) ClientAuth.REQUIRE
+                            if (trustStore.isRequireClientCertificate) ClientAuth.REQUIRE
                             else ClientAuth.OPTIONAL
                         } ?: ClientAuth.NONE
                         clientAuth(clientAuth)
@@ -245,14 +246,9 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
 
         private val log = contextLogger()
 
-        private val serverHandler = let {
-            val cacheImplementation = cfg.cache.materialize()
-            val prefix = Path.of("/").resolve(Path.of(cfg.serverPath ?: "/"))
-            ServerHandler(cacheImplementation, prefix)
-        }
+        private val cache  = cfg.cache.materialize()
 
         private val exceptionHandler = ExceptionHandler()
-        private val throttlingHandler = ThrottlingHandler(cfg)
 
         private val authenticator = when (val auth = cfg.authentication) {
             is Configuration.BasicAuthentication -> NettyHttpBasicAuthenticator(cfg.users, RoleAuthorizer())
@@ -311,7 +307,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
             cfg.connection.also { conn ->
                 val readTimeout = conn.readTimeout.toMillis()
                 val writeTimeout = conn.writeTimeout.toMillis()
-                if(readTimeout > 0 || writeTimeout > 0) {
+                if (readTimeout > 0 || writeTimeout > 0) {
                     pipeline.addLast(
                         IdleStateHandler(
                             false,
@@ -325,7 +321,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
                 val readIdleTimeout = conn.readIdleTimeout.toMillis()
                 val writeIdleTimeout = conn.writeIdleTimeout.toMillis()
                 val idleTimeout = conn.idleTimeout.toMillis()
-                if(readIdleTimeout > 0 || writeIdleTimeout > 0 || idleTimeout > 0) {
+                if (readIdleTimeout > 0 || writeIdleTimeout > 0 || idleTimeout > 0) {
                     pipeline.addLast(
                         IdleStateHandler(
                             true,
@@ -340,16 +336,19 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
             pipeline.addLast(object : ChannelInboundHandlerAdapter() {
                 override fun userEventTriggered(ctx: ChannelHandlerContext, evt: Any) {
                     if (evt is IdleStateEvent) {
-                        when(evt.state()) {
+                        when (evt.state()) {
                             IdleState.READER_IDLE -> log.debug {
                                 "Read timeout reached on channel ${ch.id().asShortText()}, closing the connection"
                             }
+
                             IdleState.WRITER_IDLE -> log.debug {
                                 "Write timeout reached on channel ${ch.id().asShortText()}, closing the connection"
                             }
+
                             IdleState.ALL_IDLE -> log.debug {
                                 "Idle timeout reached on channel ${ch.id().asShortText()}, closing the connection"
                             }
+
                             null -> throw IllegalStateException("This should never happen")
                         }
                         ctx.close()
@@ -362,34 +361,53 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
             pipeline.addLast(HttpServerCodec())
             pipeline.addLast(HttpChunkContentCompressor(1024))
             pipeline.addLast(ChunkedWriteHandler())
-            pipeline.addLast(HttpObjectAggregator(cfg.connection.maxRequestSize))
             authenticator?.let {
                 pipeline.addLast(it)
             }
-            pipeline.addLast(throttlingHandler)
+            pipeline.addLast(ThrottlingHandler(cfg))
+
+            val serverHandler = let {
+                val prefix = Path.of("/").resolve(Path.of(cfg.serverPath ?: "/"))
+                ServerHandler(cache, prefix)
+            }
             pipeline.addLast(eventExecutorGroup, serverHandler)
             pipeline.addLast(exceptionHandler)
         }
+
+        override fun close() {
+            cache.close()
+        }
     }
 
     class ServerHandle(
         httpChannelFuture: ChannelFuture,
-        private val executorGroups: Iterable<EventExecutorGroup>
+        private val executorGroups: Iterable<EventExecutorGroup>,
+        private val serverInitializer: AutoCloseable
     ) : AutoCloseable {
         private val httpChannel: Channel = httpChannelFuture.channel()
         private val closeFuture: ChannelFuture = httpChannel.closeFuture()
         private val log = contextLogger()
 
-        fun shutdown(): ChannelFuture {
+        fun shutdown(): Future<Void> {
             return httpChannel.close()
         }
 
         override fun close() {
             try {
                 closeFuture.sync()
-            } finally {
+            } catch (ex: Throwable) {
+                log.error(ex.message, ex)
+            }
+            try {
+                serverInitializer.close()
+            } catch (ex: Throwable) {
+                log.error(ex.message, ex)
+            }
             executorGroups.forEach {
+                try {
                     it.shutdownGracefully().sync()
+                } catch (ex: Throwable) {
+                    log.error(ex.message, ex)
                 }
             }
             log.info {
@@ -411,11 +429,12 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
             }
             DefaultEventExecutorGroup(Runtime.getRuntime().availableProcessors(), threadFactory)
         }
+        val serverInitializer = ServerInitializer(cfg, eventExecutorGroup)
         val bootstrap = ServerBootstrap().apply {
             // Configure the server
             group(bossGroup, workerGroup)
             channel(serverSocketChannel)
-            childHandler(ServerInitializer(cfg, eventExecutorGroup))
+            childHandler(serverInitializer)
             option(ChannelOption.SO_BACKLOG, cfg.incomingConnectionsBacklogSize)
             childOption(ChannelOption.SO_KEEPALIVE, true)
         }
@@ -427,6 +446,6 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
         log.info {
             "RemoteBuildCacheServer is listening on ${cfg.host}:${cfg.port}"
         }
-        return ServerHandle(httpChannel, setOf(bossGroup, workerGroup, eventExecutorGroup))
+        return ServerHandle(httpChannel, setOf(bossGroup, workerGroup, eventExecutorGroup), serverInitializer)
     }
 }

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/auth/Authenticator.kt

@@ -6,6 +6,7 @@ import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelInboundHandlerAdapter
 import io.netty.handler.codec.http.DefaultFullHttpResponse
 import io.netty.handler.codec.http.FullHttpResponse
+import io.netty.handler.codec.http.HttpContent
 import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpRequest
 import io.netty.handler.codec.http.HttpResponseStatus
@@ -57,6 +58,8 @@ abstract class AbstractNettyHttpAuthenticator(private val authorizer: Authorizer
             } else {
                 authorizationFailure(ctx, msg)
             }
+        } else if(msg is HttpContent) {
+            ctx.fireChannelRead(msg)
         }
     }
 

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/FileSystemCache.kt

@@ -1,13 +1,16 @@
 package net.woggioni.rbcs.server.cache
 
-import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
+import net.woggioni.jwo.JWO
 import net.woggioni.rbcs.api.Cache
-import net.woggioni.rbcs.common.ByteBufInputStream
+import net.woggioni.rbcs.api.RequestHandle
+import net.woggioni.rbcs.api.ResponseHandle
+import net.woggioni.rbcs.api.event.RequestStreamingEvent
+import net.woggioni.rbcs.api.event.ResponseStreamingEvent
+import net.woggioni.rbcs.common.ByteBufOutputStream
 import net.woggioni.rbcs.common.RBCS.digestString
 import net.woggioni.rbcs.common.contextLogger
-import net.woggioni.jwo.JWO
-import net.woggioni.jwo.LockFile
-import java.nio.channels.Channels
+import net.woggioni.rbcs.common.extractChunk
 import java.nio.channels.FileChannel
 import java.nio.file.Files
 import java.nio.file.Path
@@ -18,10 +21,8 @@ import java.security.MessageDigest
 import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.CompletableFuture
-import java.util.concurrent.atomic.AtomicReference
 import java.util.zip.Deflater
 import java.util.zip.DeflaterOutputStream
-import java.util.zip.Inflater
 import java.util.zip.InflaterInputStream
 
 class FileSystemCache(
@@ -29,7 +30,8 @@ class FileSystemCache(
     val maxAge: Duration,
     val digestAlgorithm: String?,
     val compressionEnabled: Boolean,
-    val compressionLevel: Int
+    val compressionLevel: Int,
+    val chunkSize: Int
 ) : Cache {
 
     private companion object {
@@ -41,9 +43,13 @@ class FileSystemCache(
         Files.createDirectories(root)
     }
 
-    private var nextGc = AtomicReference(Instant.now().plus(maxAge))
+    @Volatile
+    private var running = true
 
-    override fun get(key: String) = (digestAlgorithm
+    private var nextGc = Instant.now()
+
+    override fun get(key: String, responseHandle: ResponseHandle, alloc: ByteBufAllocator) {
+        (digestAlgorithm
             ?.let(MessageDigest::getInstance)
             ?.let { md ->
                 digestString(key.toByteArray(), md)
@@ -51,30 +57,57 @@ class FileSystemCache(
             root.resolve(digest).takeIf(Files::exists)
                 ?.let { file ->
                     file.takeIf(Files::exists)?.let { file ->
+                        responseHandle.handleEvent(ResponseStreamingEvent.RESPONSE_RECEIVED)
                         if (compressionEnabled) {
-                        val inflater = Inflater()
-                        Channels.newChannel(
-                            InflaterInputStream(
-                                Channels.newInputStream(
-                                    FileChannel.open(
-                                        file,
-                                        StandardOpenOption.READ
-                                    )
-                                ), inflater
-                            )
-                        )
+                            val compositeBuffer = alloc.compositeBuffer()
+                            ByteBufOutputStream(compositeBuffer).use { outputStream ->
+                                InflaterInputStream(Files.newInputStream(file)).use { inputStream ->
+                                    val ioBuffer = alloc.buffer(chunkSize)
+                                    try {
+                                        while (true) {
+                                            val read = ioBuffer.writeBytes(inputStream, chunkSize)
+                                            val last = read < 0
+                                            if (read > 0) {
+                                                ioBuffer.readBytes(outputStream, read)
+                                            }
+                                            if (last) {
+                                                compositeBuffer.retain()
+                                                outputStream.close()
+                                            }
+                                            if (compositeBuffer.readableBytes() >= chunkSize || last) {
+                                                val chunk = extractChunk(compositeBuffer, alloc)
+                                                val evt = if (last) {
+                                                    ResponseStreamingEvent.LastChunkReceived(chunk)
                                                 } else {
+                                                    ResponseStreamingEvent.ChunkReceived(chunk)
+                                                }
+                                                responseHandle.handleEvent(evt)
+                                            }
+                                            if (last) break
+                                        }
+                                    } finally {
+                                        ioBuffer.release()
+                                    }
+                                }
+                            }
+                        } else {
+                            responseHandle.handleEvent(
+                                ResponseStreamingEvent.FileReceived(
                                     FileChannel.open(file, StandardOpenOption.READ)
+                                )
+                            )
                         }
                     }
-            }.also {
-                gc()
-            }.let {
-                CompletableFuture.completedFuture(it)
+                } ?: responseHandle.handleEvent(ResponseStreamingEvent.NOT_FOUND)
         }
     }
 
-    override fun put(key: String, content: ByteBuf): CompletableFuture<Void> {
+    override fun put(
+        key: String,
+        responseHandle: ResponseHandle,
+        alloc: ByteBufAllocator
+    ): CompletableFuture<RequestHandle> {
+        try {
             (digestAlgorithm
                 ?.let(MessageDigest::getInstance)
                 ?.let { md ->
@@ -82,49 +115,87 @@ class FileSystemCache(
                 } ?: key).let { digest ->
                 val file = root.resolve(digest)
                 val tmpFile = Files.createTempFile(root, null, ".tmp")
-            try {
-                Files.newOutputStream(tmpFile).let {
+                val stream = Files.newOutputStream(tmpFile).let {
                     if (compressionEnabled) {
                         val deflater = Deflater(compressionLevel)
                         DeflaterOutputStream(it, deflater)
                     } else {
                         it
                     }
-                }.use {
-                    JWO.copy(ByteBufInputStream(content), it)
                 }
+                return CompletableFuture.completedFuture(object : RequestHandle {
+                    override fun handleEvent(evt: RequestStreamingEvent) {
+                        try {
+                            when (evt) {
+                                is RequestStreamingEvent.LastChunkReceived -> {
+                                    evt.chunk.readBytes(stream, evt.chunk.readableBytes())
+                                    stream.close()
                                     Files.move(tmpFile, file, StandardCopyOption.ATOMIC_MOVE)
-            } catch (t: Throwable) {
-                Files.delete(tmpFile)
-                throw t
+                                    responseHandle.handleEvent(ResponseStreamingEvent.RESPONSE_RECEIVED)
                                 }
-        }.also {
+
+                                is RequestStreamingEvent.ChunkReceived -> {
+                                    evt.chunk.readBytes(stream, evt.chunk.readableBytes())
+                                }
+
+                                is RequestStreamingEvent.ExceptionCaught -> {
+                                    Files.delete(tmpFile)
+                                    stream.close()
+                                }
+                            }
+                        } catch (ex: Throwable) {
+                            responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(ex))
+                        }
+                    }
+                })
+            }
+        } catch (ex: Throwable) {
+            responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(ex))
+            return CompletableFuture.failedFuture(ex)
+        }
+    }
+
+    private val garbageCollector = Thread.ofVirtual().name("file-system-cache-gc").start {
+        while (running) {
             gc()
         }
-        return CompletableFuture.completedFuture(null)
     }
 
     private fun gc() {
         val now = Instant.now()
-        val oldValue = nextGc.getAndSet(now.plus(maxAge))
-        if (oldValue < now) {
-            actualGc(now)
+        if (nextGc < now) {
+            val oldestEntry = actualGc(now)
+            nextGc = (oldestEntry ?: now).plus(maxAge)
         }
+        Thread.sleep(minOf(Duration.between(now, nextGc), Duration.ofSeconds(1)))
     }
 
-    @Synchronized
-    private fun actualGc(now: Instant) {
-        Files.list(root).filter {
+    /**
+     * Returns the creation timestamp of the oldest cache entry (if any)
+     */
+    private fun actualGc(now: Instant): Instant? {
+        var result: Instant? = null
+        Files.list(root)
+            .filter { path ->
+                JWO.splitExtension(path)
+                    .map { it._2 }
+                    .map { it != ".tmp" }
+                    .orElse(true)
+            }
+            .filter {
                 val creationTimeStamp = Files.readAttributes(it, BasicFileAttributes::class.java)
                     .creationTime()
                     .toInstant()
+                if (result == null || creationTimeStamp < result) {
+                    result = creationTimeStamp
+                }
                 now > creationTimeStamp.plus(maxAge)
-        }.forEach { file ->
-            LockFile.acquire(file, false).use {
-                Files.delete(file)
-            }
-        }
+            }.forEach(Files::delete)
+        return result
     }
 
-    override fun close() {}
+    override fun close() {
+        running = false
+        garbageCollector.join()
+    }
 }

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/FileSystemCacheConfiguration.kt

@@ -12,13 +12,15 @@ data class FileSystemCacheConfiguration(
     val digestAlgorithm : String?,
     val compressionEnabled: Boolean,
     val compressionLevel: Int,
+    val chunkSize: Int,
 ) : Configuration.Cache {
     override fun materialize() = FileSystemCache(
         root ?: Application.builder("rbcs").build().computeCacheDirectory(),
         maxAge,
         digestAlgorithm,
         compressionEnabled,
-        compressionLevel
+        compressionLevel,
+        chunkSize,
     )
 
     override fun getNamespaceURI() = RBCS.RBCS_NAMESPACE_URI

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/FileSystemCacheProvider.kt

@@ -31,13 +31,17 @@ class FileSystemCacheProvider : CacheProvider<FileSystemCacheConfiguration> {
             ?.let(String::toInt)
             ?: Deflater.DEFAULT_COMPRESSION
         val digestAlgorithm = el.renderAttribute("digest") ?: "MD5"
+        val chunkSize = el.renderAttribute("chunk-size")
+            ?.let(Integer::decode)
+            ?: 0x4000
 
         return FileSystemCacheConfiguration(
             path,
             maxAge,
             digestAlgorithm,
             enableCompression,
-            compressionLevel
+            compressionLevel,
+            chunkSize
         )
     }
 
@@ -57,6 +61,7 @@ class FileSystemCacheProvider : CacheProvider<FileSystemCacheConfiguration> {
             }?.let {
                 attr("compression-level", it.toString())
             }
+            attr("chunk-size", chunkSize.toString())
         }
         result
     }

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCache.kt

@@ -1,31 +1,36 @@
 package net.woggioni.rbcs.server.cache
 
 import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
 import net.woggioni.rbcs.api.Cache
-import net.woggioni.rbcs.common.ByteBufInputStream
+import net.woggioni.rbcs.api.RequestHandle
+import net.woggioni.rbcs.api.ResponseHandle
+import net.woggioni.rbcs.api.event.RequestStreamingEvent
+import net.woggioni.rbcs.api.event.ResponseStreamingEvent
 import net.woggioni.rbcs.common.ByteBufOutputStream
 import net.woggioni.rbcs.common.RBCS.digestString
 import net.woggioni.rbcs.common.contextLogger
-import net.woggioni.jwo.JWO
-import java.nio.channels.Channels
+import net.woggioni.rbcs.common.extractChunk
 import java.security.MessageDigest
 import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.PriorityBlockingQueue
+import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicLong
 import java.util.zip.Deflater
 import java.util.zip.DeflaterOutputStream
 import java.util.zip.Inflater
-import java.util.zip.InflaterInputStream
+import java.util.zip.InflaterOutputStream
 
 class InMemoryCache(
-    val maxAge: Duration,
-    val maxSize: Long,
-    val digestAlgorithm: String?,
-    val compressionEnabled: Boolean,
-    val compressionLevel: Int
+    private val maxAge: Duration,
+    private val maxSize: Long,
+    private val digestAlgorithm: String?,
+    private val compressionEnabled: Boolean,
+    private val compressionLevel: Int,
+    private val chunkSize : Int
 ) : Cache {
 
     companion object {
@@ -36,44 +41,41 @@ class InMemoryCache(
     private val size = AtomicLong()
     private val map = ConcurrentHashMap<String, ByteBuf>()
 
-    private class RemovalQueueElement(val key: String, val value : ByteBuf, val expiry : Instant) : Comparable<RemovalQueueElement> {
+    private class RemovalQueueElement(val key: String, val value: ByteBuf, val expiry: Instant) :
+        Comparable<RemovalQueueElement> {
         override fun compareTo(other: RemovalQueueElement) = expiry.compareTo(other.expiry)
     }
 
     private val removalQueue = PriorityBlockingQueue<RemovalQueueElement>()
 
+    @Volatile
     private var running = true
-    private val garbageCollector = Thread {
-        while(true) {
-            val el = removalQueue.take()
+
+    private val garbageCollector = Thread.ofVirtual().name("in-memory-cache-gc").start {
+        while (running) {
+            val el = removalQueue.poll(1, TimeUnit.SECONDS) ?: continue
             val buf = el.value
             val now = Instant.now()
-            if(now > el.expiry) {
+            if (now > el.expiry) {
                 val removed = map.remove(el.key, buf)
-                if(removed) {
+                if (removed) {
                     updateSizeAfterRemoval(buf)
                     //Decrease the reference count for map
                     buf.release()
                 }
-                //Decrease the reference count for removalQueue
-                buf.release()
             } else {
                 removalQueue.put(el)
                 Thread.sleep(minOf(Duration.between(now, el.expiry), Duration.ofSeconds(1)))
             }
         }
-    }.apply {
-        start()
     }
 
-    private fun removeEldest() : Long {
-        while(true) {
+    private fun removeEldest(): Long {
+        while (true) {
             val el = removalQueue.take()
             val buf = el.value
             val removed = map.remove(el.key, buf)
-            //Decrease the reference count for removalQueue
-            buf.release()
-            if(removed) {
+            if (removed) {
                 val newSize = updateSizeAfterRemoval(buf)
                 //Decrease the reference count for map
                 buf.release()
@@ -82,8 +84,8 @@ class InMemoryCache(
         }
     }
 
-    private fun updateSizeAfterRemoval(removed: ByteBuf) : Long {
-        return size.updateAndGet { currentSize : Long ->
+    private fun updateSizeAfterRemoval(removed: ByteBuf): Long {
+        return size.updateAndGet { currentSize: Long ->
             currentSize - removed.readableBytes()
         }
     }
@@ -93,7 +95,8 @@ class InMemoryCache(
         garbageCollector.join()
     }
 
-    override fun get(key: String) =
+    override fun get(key: String, responseHandle: ResponseHandle, alloc: ByteBufAllocator) {
+        try {
             (digestAlgorithm
                 ?.let(MessageDigest::getInstance)
                 ?.let { md ->
@@ -103,48 +106,103 @@ class InMemoryCache(
                     map[digest]
                         ?.let { value ->
                             val copy = value.retainedDuplicate()
-                    copy.touch("This has to be released by the caller of the cache")
+                            responseHandle.handleEvent(ResponseStreamingEvent.RESPONSE_RECEIVED)
+                            val output = alloc.compositeBuffer()
                             if (compressionEnabled) {
+                                try {
+                                    val stream = ByteBufOutputStream(output).let {
                                         val inflater = Inflater()
-                        Channels.newChannel(InflaterInputStream(ByteBufInputStream(copy), inflater))
+                                        InflaterOutputStream(it, inflater)
+                                    }
+                                    stream.use { os ->
+                                        var readable = copy.readableBytes()
+                                        while (true) {
+                                            copy.readBytes(os, chunkSize.coerceAtMost(readable))
+                                            readable = copy.readableBytes()
+                                            val last = readable == 0
+                                            if (last) stream.flush()
+                                            if (output.readableBytes() >= chunkSize || last) {
+                                                val chunk = extractChunk(output, alloc)
+                                                val evt = if (last) {
+                                                    ResponseStreamingEvent.LastChunkReceived(chunk)
                                                 } else {
-                        Channels.newChannel(ByteBufInputStream(copy))
+                                                    ResponseStreamingEvent.ChunkReceived(chunk)
+                                                }
+                                                responseHandle.handleEvent(evt)
+                                            }
+                                            if (last) break
                                         }
                                     }
-            }.let {
-                CompletableFuture.completedFuture(it)
+                                } finally {
+                                    copy.release()
+                                }
+                            } else {
+                                responseHandle.handleEvent(
+                                    ResponseStreamingEvent.LastChunkReceived(copy)
+                                )
+                            }
+                        } ?: responseHandle.handleEvent(ResponseStreamingEvent.NOT_FOUND)
+                }
+        } catch (ex: Throwable) {
+            responseHandle.handleEvent(ResponseStreamingEvent.ExceptionCaught(ex))
+        }
     }
 
-    override fun put(key: String, content: ByteBuf) =
+    override fun put(
+        key: String,
+        responseHandle: ResponseHandle,
+        alloc: ByteBufAllocator
+    ): CompletableFuture<RequestHandle> {
+        return CompletableFuture.completedFuture(object : RequestHandle {
+            val buf = alloc.heapBuffer()
+            val stream = ByteBufOutputStream(buf).let {
+                if (compressionEnabled) {
+                    val deflater = Deflater(compressionLevel)
+                    DeflaterOutputStream(it, deflater)
+                } else {
+                    it
+                }
+            }
+
+            override fun handleEvent(evt: RequestStreamingEvent) {
+                when (evt) {
+                    is RequestStreamingEvent.ChunkReceived -> {
+                        evt.chunk.readBytes(stream, evt.chunk.readableBytes())
+                        if (evt is RequestStreamingEvent.LastChunkReceived) {
                             (digestAlgorithm
                                 ?.let(MessageDigest::getInstance)
                                 ?.let { md ->
                                     digestString(key.toByteArray(), md)
-            } ?: key).let { digest ->
-            content.retain()
-            val value = if (compressionEnabled) {
-                val deflater = Deflater(compressionLevel)
-                val buf = content.alloc().buffer()
-                buf.retain()
-                DeflaterOutputStream(ByteBufOutputStream(buf), deflater).use { outputStream ->
-                    ByteBufInputStream(content).use { inputStream ->
-                        JWO.copy(inputStream, outputStream)
-                    }
-                }
-                buf
-            } else {
-                content
-            }
-            val old = map.put(digest, value)
-            val delta = value.readableBytes() - (old?.readableBytes() ?: 0)
+                                } ?: key
+                                    ).let { digest ->
+                                    val oldSize = map.put(digest, buf.retain())?.let { old ->
+                                        val result = old.readableBytes()
+                                        old.release()
+                                        result
+                                    } ?: 0
+                                    val delta = buf.readableBytes() - oldSize
                                     var newSize = size.updateAndGet { currentSize : Long ->
                                         currentSize + delta
                                     }
-            removalQueue.put(RemovalQueueElement(digest, value.retain(), Instant.now().plus(maxAge)))
+                                    removalQueue.put(RemovalQueueElement(digest, buf, Instant.now().plus(maxAge)))
                                     while(newSize > maxSize) {
                                         newSize = removeEldest()
                                     }
-        }.let {
-            CompletableFuture.completedFuture<Void>(null)
+                                    stream.close()
+                                    responseHandle.handleEvent(ResponseStreamingEvent.RESPONSE_RECEIVED)
+                                }
+                        }
+                    }
+
+                    is RequestStreamingEvent.ExceptionCaught -> {
+                        stream.close()
+                    }
+
+                    else -> {
+
+                    }
+                }
+            }
+        })
     }
 }

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCacheConfiguration.kt

@@ -10,13 +10,15 @@ data class InMemoryCacheConfiguration(
     val digestAlgorithm : String?,
     val compressionEnabled: Boolean,
     val compressionLevel: Int,
+    val chunkSize : Int
 ) : Configuration.Cache {
     override fun materialize() = InMemoryCache(
         maxAge,
         maxSize,
         digestAlgorithm,
         compressionEnabled,
-        compressionLevel
+        compressionLevel,
+        chunkSize
     )
 
     override fun getNamespaceURI() = RBCS.RBCS_NAMESPACE_URI

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCacheProvider.kt

@@ -31,13 +31,16 @@ class InMemoryCacheProvider : CacheProvider<InMemoryCacheConfiguration> {
             ?.let(String::toInt)
             ?: Deflater.DEFAULT_COMPRESSION
         val digestAlgorithm = el.renderAttribute("digest") ?: "MD5"
-
+        val chunkSize = el.renderAttribute("chunk-size")
+            ?.let(Integer::decode)
+            ?: 0x4000
         return InMemoryCacheConfiguration(
             maxAge,
             maxSize,
             digestAlgorithm,
             enableCompression,
-            compressionLevel
+            compressionLevel,
+            chunkSize
         )
     }
 
@@ -57,6 +60,7 @@ class InMemoryCacheProvider : CacheProvider<InMemoryCacheConfiguration> {
             }?.let {
                 attr("compression-level", it.toString())
             }
+            attr("chunk-size", chunkSize.toString())
         }
         result
     }

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/configuration/Parser.kt

@@ -124,7 +124,7 @@ object Parser {
                     val writeIdleTimeout = child.renderAttribute("write-idle-timeout")
                         ?.let(Duration::parse) ?: Duration.of(60, ChronoUnit.SECONDS)
                     val maxRequestSize = child.renderAttribute("max-request-size")
-                        ?.let(String::toInt) ?: 67108864
+                        ?.let(Integer::decode) ?: 0x4000000
                     connection = Configuration.Connection(
                         readTimeout,
                         writeTimeout,

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/exception/ExceptionHandler.kt

@@ -17,6 +17,8 @@ import net.woggioni.rbcs.api.exception.CacheException
 import net.woggioni.rbcs.api.exception.ContentTooLargeException
 import net.woggioni.rbcs.common.contextLogger
 import net.woggioni.rbcs.common.debug
+import java.net.SocketException
+import javax.net.ssl.SSLException
 import javax.net.ssl.SSLPeerUnverifiedException
 
 @ChannelHandler.Sharable
@@ -50,7 +52,12 @@ class ExceptionHandler : ChannelDuplexHandler() {
     override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
         when (cause) {
             is DecoderException -> {
-                log.error(cause.message, cause)
+                log.debug(cause.message, cause)
+                ctx.close()
+            }
+
+            is SocketException -> {
+                log.debug(cause.message, cause)
                 ctx.close()
             }
 
@@ -59,10 +66,16 @@ class ExceptionHandler : ChannelDuplexHandler() {
                     .addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
             }
 
+            is SSLException -> {
+                log.debug(cause.message, cause)
+                ctx.close()
+            }
+
             is ContentTooLargeException -> {
                 ctx.writeAndFlush(TOO_BIG.retainedDuplicate())
                     .addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
             }
+
             is ReadTimeoutException -> {
                 log.debug {
                     val channelId = ctx.channel().id().asShortText()
@@ -70,6 +83,7 @@ class ExceptionHandler : ChannelDuplexHandler() {
                 }
                 ctx.close()
             }
+
             is WriteTimeoutException -> {
                 log.debug {
                     val channelId = ctx.channel().id().asShortText()
@@ -77,11 +91,13 @@ class ExceptionHandler : ChannelDuplexHandler() {
                 }
                 ctx.close()
             }
+
             is CacheException -> {
                 log.error(cause.message, cause)
                 ctx.writeAndFlush(NOT_AVAILABLE.retainedDuplicate())
                     .addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
             }
+
             else -> {
                 log.error(cause.message, cause)
                 ctx.writeAndFlush(SERVER_ERROR.retainedDuplicate())

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/handler/ServerHandler.kt

@@ -2,34 +2,66 @@ package net.woggioni.rbcs.server.handler
 
 import io.netty.buffer.Unpooled
 import io.netty.channel.ChannelFutureListener
-import io.netty.channel.ChannelHandler
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.DefaultFileRegion
 import io.netty.channel.SimpleChannelInboundHandler
 import io.netty.handler.codec.http.DefaultFullHttpResponse
+import io.netty.handler.codec.http.DefaultHttpContent
 import io.netty.handler.codec.http.DefaultHttpResponse
-import io.netty.handler.codec.http.FullHttpRequest
+import io.netty.handler.codec.http.DefaultLastHttpContent
+import io.netty.handler.codec.http.HttpContent
 import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpHeaderValues
 import io.netty.handler.codec.http.HttpMethod
+import io.netty.handler.codec.http.HttpObject
+import io.netty.handler.codec.http.HttpRequest
 import io.netty.handler.codec.http.HttpResponseStatus
 import io.netty.handler.codec.http.HttpUtil
 import io.netty.handler.codec.http.LastHttpContent
-import io.netty.handler.stream.ChunkedNioStream
 import net.woggioni.rbcs.api.Cache
+import net.woggioni.rbcs.api.RequestHandle
+import net.woggioni.rbcs.api.ResponseHandle
+import net.woggioni.rbcs.api.event.RequestStreamingEvent
+import net.woggioni.rbcs.api.event.ResponseStreamingEvent
 import net.woggioni.rbcs.common.contextLogger
+import net.woggioni.rbcs.common.debug
 import net.woggioni.rbcs.server.debug
 import net.woggioni.rbcs.server.warn
-import java.nio.channels.FileChannel
 import java.nio.file.Path
+import java.util.concurrent.CompletableFuture
 
-@ChannelHandler.Sharable
 class ServerHandler(private val cache: Cache, private val serverPrefix: Path) :
-    SimpleChannelInboundHandler<FullHttpRequest>() {
+    SimpleChannelInboundHandler<HttpObject>() {
 
     private val log = contextLogger()
 
-    override fun channelRead0(ctx: ChannelHandlerContext, msg: FullHttpRequest) {
+    override fun channelRead0(ctx: ChannelHandlerContext, msg: HttpObject) {
+        when(msg) {
+            is HttpRequest -> handleRequest(ctx, msg)
+            is HttpContent -> handleContent(msg)
+        }
+    }
+
+    private var requestHandle : CompletableFuture<RequestHandle?> = CompletableFuture.completedFuture(null)
+
+    private fun handleContent(content : HttpContent) {
+            content.retain()
+            requestHandle.thenAccept { handle ->
+                handle?.let {
+                    val evt = if(content is LastHttpContent) {
+                        RequestStreamingEvent.LastChunkReceived(content.content())
+
+                    } else {
+                        RequestStreamingEvent.ChunkReceived(content.content())
+                    }
+                    it.handleEvent(evt)
+                    content.release()
+                } ?: content.release()
+            }
+        }
+
+
+    private fun handleRequest(ctx : ChannelHandlerContext, msg : HttpRequest) {
         val keepAlive: Boolean = HttpUtil.isKeepAlive(msg)
         val method = msg.method()
         if (method === HttpMethod.GET) {
@@ -42,24 +74,44 @@ class ServerHandler(private val cache: Cache, private val serverPrefix: Path) :
                 return
             }
             if (serverPrefix == prefix) {
-                cache.get(key).thenApply { channel ->
-                    if(channel != null) {
-                        log.debug(ctx) {
-                            "Cache hit for key '$key'"
-                        }
+                val responseHandle = ResponseHandle { evt ->
+                    when (evt) {
+                        is ResponseStreamingEvent.ResponseReceived -> {
                             val response = DefaultHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK)
                             response.headers()[HttpHeaderNames.CONTENT_TYPE] = HttpHeaderValues.APPLICATION_OCTET_STREAM
                             if (!keepAlive) {
                                 response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
-                            response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.IDENTITY)
                             } else {
                                 response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE)
-                            response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED)
                             }
-                        ctx.write(response)
-                        when (channel) {
-                            is FileChannel -> {
-                                val content = DefaultFileRegion(channel, 0, channel.size())
+                            response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED)
+                            ctx.writeAndFlush(response)
+                        }
+
+                        is ResponseStreamingEvent.LastChunkReceived -> {
+                            val channelFuture = ctx.writeAndFlush(DefaultLastHttpContent(evt.chunk))
+                            if (!keepAlive) {
+                                channelFuture
+                                    .addListener(ChannelFutureListener.CLOSE)
+                            }
+                        }
+
+                        is ResponseStreamingEvent.ChunkReceived -> {
+                            ctx.writeAndFlush(DefaultHttpContent(evt.chunk))
+                        }
+
+                        is ResponseStreamingEvent.ExceptionCaught -> {
+                            ctx.fireExceptionCaught(evt.exception)
+                        }
+
+                        is ResponseStreamingEvent.NotFound -> {
+                            val response = DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.NOT_FOUND)
+                            response.headers()[HttpHeaderNames.CONTENT_LENGTH] = 0
+                            ctx.writeAndFlush(response)
+                        }
+
+                        is ResponseStreamingEvent.FileReceived -> {
+                            val content = DefaultFileRegion(evt.file, 0, evt.file.size())
                             if (keepAlive) {
                                 ctx.write(content)
                                 ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT.retainedDuplicate())
@@ -68,28 +120,9 @@ class ServerHandler(private val cache: Cache, private val serverPrefix: Path) :
                                     .addListener(ChannelFutureListener.CLOSE)
                             }
                         }
-                            else -> {
-                                val content = ChunkedNioStream(channel)
-                                if (keepAlive) {
-                                    ctx.write(content).addListener {
-                                        content.close()
-                                    }
-                                    ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT.retainedDuplicate())
-                                } else {
-                                    ctx.writeAndFlush(content)
-                                        .addListener(ChannelFutureListener.CLOSE)
                     }
                 }
-                        }
-                    } else {
-                        log.debug(ctx) {
-                            "Cache miss for key '$key'"
-                        }
-                        val response = DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.NOT_FOUND)
-                        response.headers()[HttpHeaderNames.CONTENT_LENGTH] = 0
-                        ctx.writeAndFlush(response)
-                    }
-                }.whenComplete { _, ex -> ex?.let(ctx::fireExceptionCaught) }
+                cache.get(key, responseHandle, ctx.alloc())
             } else {
                 log.warn(ctx) {
                     "Got request for unhandled path '${msg.uri()}'"
@@ -107,15 +140,32 @@ class ServerHandler(private val cache: Cache, private val serverPrefix: Path) :
                 log.debug(ctx) {
                     "Added value for key '$key' to build cache"
                 }
-                cache.put(key, msg.content()).thenRun {
+                val responseHandle = ResponseHandle { evt ->
+                    when (evt) {
+                        is ResponseStreamingEvent.ResponseReceived -> {
                             val response = DefaultFullHttpResponse(
                                 msg.protocolVersion(), HttpResponseStatus.CREATED,
                                 Unpooled.copiedBuffer(key.toByteArray())
                             )
                             response.headers()[HttpHeaderNames.CONTENT_LENGTH] = response.content().readableBytes()
                             ctx.writeAndFlush(response)
-                }.whenComplete { _, ex ->
+                            this.requestHandle = CompletableFuture.completedFuture(null)
+                        }
+                        is ResponseStreamingEvent.ChunkReceived -> {
+                            evt.chunk.release()
+                        }
+                        is ResponseStreamingEvent.ExceptionCaught -> {
+                            ctx.fireExceptionCaught(evt.exception)
+                        }
+                        else -> {}
+                    }
+                }
+
+                this.requestHandle = cache.put(key, responseHandle, ctx.alloc()).exceptionally { ex ->
                     ctx.fireExceptionCaught(ex)
+                    null
+                }.also {
+                    log.debug { "Replacing request handle with $it"}
                 }
             } else {
                 log.warn(ctx) {
@@ -125,9 +175,12 @@ class ServerHandler(private val cache: Cache, private val serverPrefix: Path) :
                 response.headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
                 ctx.writeAndFlush(response)
             }
-        } else if(method == HttpMethod.TRACE) {
+        } else if (method == HttpMethod.TRACE) {
             val replayedRequestHead = ctx.alloc().buffer()
-            replayedRequestHead.writeCharSequence("TRACE ${Path.of(msg.uri())} ${msg.protocolVersion().text()}\r\n", Charsets.US_ASCII)
+            replayedRequestHead.writeCharSequence(
+                "TRACE ${Path.of(msg.uri())} ${msg.protocolVersion().text()}\r\n",
+                Charsets.US_ASCII
+            )
             msg.headers().forEach { (key, value) ->
                 replayedRequestHead.apply {
                     writeCharSequence(key, Charsets.US_ASCII)
@@ -137,16 +190,24 @@ class ServerHandler(private val cache: Cache, private val serverPrefix: Path) :
                 }
             }
             replayedRequestHead.writeCharSequence("\r\n", Charsets.US_ASCII)
-            val requestBody = msg.content()
-            requestBody.retain()
-            val responseBody = ctx.alloc().compositeBuffer(2).apply {
-                addComponents(true, replayedRequestHead)
-                addComponents(true, requestBody)
+            this.requestHandle = CompletableFuture.completedFuture(RequestHandle { evt ->
+                when(evt) {
+                    is RequestStreamingEvent.LastChunkReceived -> {
+                        ctx.writeAndFlush(DefaultLastHttpContent(evt.chunk.retain()))
+                        this.requestHandle = CompletableFuture.completedFuture(null)
                     }
-            val response = DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK, responseBody)
+                    is RequestStreamingEvent.ChunkReceived -> ctx.writeAndFlush(DefaultHttpContent(evt.chunk.retain()))
+                    is RequestStreamingEvent.ExceptionCaught -> ctx.fireExceptionCaught(evt.exception)
+                    else -> {
+
+                    }
+                }
+            }).also {
+                log.debug { "Replacing request handle with $it"}
+            }
+            val response = DefaultHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK)
             response.headers().apply {
                 set(HttpHeaderNames.CONTENT_TYPE, "message/http")
-                set(HttpHeaderNames.CONTENT_LENGTH, responseBody.readableBytes())
             }
             ctx.writeAndFlush(response)
         } else {
@@ -158,4 +219,11 @@ class ServerHandler(private val cache: Cache, private val serverPrefix: Path) :
             ctx.writeAndFlush(response)
         }
     }
+
+    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
+        requestHandle.thenAccept { handle ->
+            handle?.handleEvent(RequestStreamingEvent.ExceptionCaught(cause))
+        }
+        super.exceptionCaught(ctx, cause)
+    }
 }

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/throttling/ThrottlingHandler.kt

@@ -1,10 +1,11 @@
 package net.woggioni.rbcs.server.throttling
 
-import io.netty.channel.ChannelHandler.Sharable
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelInboundHandlerAdapter
 import io.netty.handler.codec.http.DefaultFullHttpResponse
+import io.netty.handler.codec.http.HttpContent
 import io.netty.handler.codec.http.HttpHeaderNames
+import io.netty.handler.codec.http.HttpRequest
 import io.netty.handler.codec.http.HttpResponseStatus
 import io.netty.handler.codec.http.HttpVersion
 import net.woggioni.rbcs.api.Configuration
@@ -18,15 +19,19 @@ import java.time.temporal.ChronoUnit
 import java.util.concurrent.TimeUnit
 
 
-@Sharable
-class ThrottlingHandler(cfg: Configuration) :
-    ChannelInboundHandlerAdapter() {
+class ThrottlingHandler(cfg: Configuration) : ChannelInboundHandlerAdapter() {
 
+    private companion object {
+        @JvmStatic
         private val log = contextLogger()
+    }
+
     private val bucketManager = BucketManager.from(cfg)
 
     private val connectionConfiguration = cfg.connection
 
+    private var queuedContent : MutableList<HttpContent>? = null
+
     /**
      * If the suggested waiting time from the bucket is lower than this
      * amount, then the server will simply wait by itself before sending a response
@@ -38,7 +43,10 @@ class ThrottlingHandler(cfg: Configuration) :
         connectionConfiguration.writeIdleTimeout
     ).dividedBy(2)
 
+
+
     override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
+        if(msg is HttpRequest) {
             val buckets = mutableListOf<Bucket>()
             val user = ctx.channel().attr(RemoteBuildCacheServer.userAttribute).get()
             if (user != null) {
@@ -54,13 +62,19 @@ class ThrottlingHandler(cfg: Configuration) :
                 bucketManager.getBucketByAddress(ctx.channel().remoteAddress() as InetSocketAddress)?.let(buckets::add)
             }
             if (buckets.isEmpty()) {
-            return super.channelRead(ctx, msg)
+                super.channelRead(ctx, msg)
             } else {
                 handleBuckets(buckets, ctx, msg, true)
             }
+            ctx.channel().id()
+        } else if(msg is HttpContent) {
+            queuedContent?.add(msg) ?: super.channelRead(ctx, msg)
+        } else {
+            super.channelRead(ctx, msg)
+        }
     }
 
-    private fun handleBuckets(buckets : List<Bucket>, ctx : ChannelHandlerContext, msg : Any, delayResponse : Boolean) {
+    private fun handleBuckets(buckets: List<Bucket>, ctx: ChannelHandlerContext, msg: Any, delayResponse: Boolean) {
         var nextAttempt = -1L
         for (bucket in buckets) {
             val bucketNextAttempt = bucket.removeTokensWithEstimate(1)
@@ -68,12 +82,18 @@ class ThrottlingHandler(cfg: Configuration) :
                 nextAttempt = bucketNextAttempt
             }
         }
-        if(nextAttempt < 0) {
+        if (nextAttempt < 0) {
             super.channelRead(ctx, msg)
-            return
+            queuedContent?.let {
+                for(content in it) {
+                    super.channelRead(ctx, content)
                 }
+                queuedContent = null
+            }
+        } else {
             val waitDuration = Duration.of(LongMath.ceilDiv(nextAttempt, 100_000_000L) * 100L, ChronoUnit.MILLIS)
             if (delayResponse && waitDuration < waitThreshold) {
+                this.queuedContent = mutableListOf()
                 ctx.executor().schedule({
                     handleBuckets(buckets, ctx, msg, false)
                 }, waitDuration.toMillis(), TimeUnit.MILLISECONDS)
@@ -81,6 +101,7 @@ class ThrottlingHandler(cfg: Configuration) :
                 sendThrottledResponse(ctx, waitDuration)
             }
         }
+    }
 
     private fun sendThrottledResponse(ctx: ChannelHandlerContext, retryAfter: Duration) {
         val response = DefaultFullHttpResponse(

rbcs-server/src/main/resources/net/woggioni/rbcs/server/schema/rbcs.xsd

@@ -39,7 +39,7 @@
         <xs:attribute name="idle-timeout" type="xs:duration" use="optional" default="PT30S"/>
         <xs:attribute name="read-idle-timeout" type="xs:duration" use="optional" default="PT60S"/>
         <xs:attribute name="write-idle-timeout" type="xs:duration" use="optional" default="PT60S"/>
-        <xs:attribute name="max-request-size" type="xs:unsignedInt" use="optional" default="67108864"/>
+        <xs:attribute name="max-request-size" type="rbcs:byteSize" use="optional" default="0x4000000"/>
     </xs:complexType>
 
     <xs:complexType name="eventExecutorType">
@@ -52,10 +52,11 @@
         <xs:complexContent>
             <xs:extension base="rbcs:cacheType">
                 <xs:attribute name="max-age" type="xs:duration" default="P1D"/>
-                <xs:attribute name="max-size" type="xs:token" default="0x1000000"/>
+                <xs:attribute name="max-size" type="rbcs:byteSize" default="0x1000000"/>
                 <xs:attribute name="digest" type="xs:token" default="MD5"/>
                 <xs:attribute name="enable-compression" type="xs:boolean" default="true"/>
                 <xs:attribute name="compression-level" type="xs:byte" default="-1"/>
+                <xs:attribute name="chunk-size" type="rbcs:byteSize" default="0x4000"/>
             </xs:extension>
         </xs:complexContent>
     </xs:complexType>
@@ -68,6 +69,7 @@
                 <xs:attribute name="digest" type="xs:token" default="MD5"/>
                 <xs:attribute name="enable-compression" type="xs:boolean" default="true"/>
                 <xs:attribute name="compression-level" type="xs:byte" default="-1"/>
+                <xs:attribute name="chunk-size" type="rbcs:byteSize" default="0x4000"/>
             </xs:extension>
         </xs:complexContent>
     </xs:complexType>
@@ -220,5 +222,10 @@
         <xs:attribute name="port" type="xs:unsignedShort" use="required"/>
     </xs:complexType>
 
+    <xs:simpleType name="byteSize">
+        <xs:restriction base="xs:token">
+            <xs:pattern value="(0x[a-f0-9]+|[0-9]+)"/>
+        </xs:restriction>
+    </xs:simpleType>
 
 </xs:schema>

rbcs-server/src/test/kotlin/net/woggioni/rbcs/server/test/AbstractBasicAuthServerTest.kt

@@ -47,11 +47,13 @@ abstract class AbstractBasicAuthServerTest : AbstractServerTest() {
             ),
             users.asSequence().map { it.name to it}.toMap(),
             sequenceOf(writersGroup, readersGroup).map { it.name to it}.toMap(),
-            FileSystemCacheConfiguration(this.cacheDir,
+            FileSystemCacheConfiguration(
+                this.cacheDir,
                 maxAge = Duration.ofSeconds(3600 * 24),
                 digestAlgorithm = "MD5",
                 compressionLevel = Deflater.DEFAULT_COMPRESSION,
-                compressionEnabled = false
+                compressionEnabled = false,
+                chunkSize = 0x1000
             ),
             Configuration.BasicAuthentication(),
             null,

rbcs-server/src/test/kotlin/net/woggioni/rbcs/server/test/AbstractTlsServerTest.kt

@@ -156,7 +156,8 @@ abstract class AbstractTlsServerTest : AbstractServerTest() {
                 maxAge = Duration.ofSeconds(3600 * 24),
                 compressionEnabled = true,
                 compressionLevel = Deflater.DEFAULT_COMPRESSION,
-                digestAlgorithm = "MD5"
+                digestAlgorithm = "MD5",
+                chunkSize = 0x1000
             ),
 //            InMemoryCacheConfiguration(
 //                maxAge = Duration.ofSeconds(3600 * 24),

rbcs-server/src/test/kotlin/net/woggioni/rbcs/server/test/BasicAuthServerTest.kt

@@ -86,7 +86,7 @@ class BasicAuthServerTest : AbstractBasicAuthServerTest() {
     @Test
     @Order(4)
     fun putAsAWriterUser() {
-        val client: HttpClient = HttpClient.newHttpClient()
+        val client: HttpClient = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build()
 
         val (key, value) = keyValuePair
         val user = cfg.users.values.find {

rbcs-server/src/test/kotlin/net/woggioni/rbcs/server/test/NoAuthServerTest.kt

@@ -52,7 +52,8 @@ class NoAuthServerTest : AbstractServerTest() {
                 compressionEnabled = true,
                 digestAlgorithm = "MD5",
                 compressionLevel = Deflater.DEFAULT_COMPRESSION,
-                maxSize = 0x1000000
+                maxSize = 0x1000000,
+                chunkSize = 0x1000
             ),
             null,
             null,
@@ -80,7 +81,7 @@ class NoAuthServerTest : AbstractServerTest() {
     @Test
     @Order(1)
     fun putWithNoAuthorizationHeader() {
-        val client: HttpClient = HttpClient.newHttpClient()
+        val client: HttpClient = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build()
         val (key, value) = keyValuePair
 
         val requestBuilder = newRequestBuilder(key)

rbcs-server/src/test/resources/net/woggioni/rbcs/server/test/valid/rbcs-default.xml

@@ -11,7 +11,7 @@
             idle-timeout="PT30M"
             max-request-size="101325"/>
     <event-executor use-virtual-threads="false"/>
-    <cache xs:type="rbcs:fileSystemCacheType" path="/tmp/rbcs" max-age="P7D"/>
+    <cache xs:type="rbcs:fileSystemCacheType" path="/tmp/rbcs" max-age="P7D" chunk-size="0xa910"/>
     <authentication>
         <none/>
     </authentication>

rbcs-server/src/test/resources/net/woggioni/rbcs/server/test/valid/rbcs-memcached-tls.xml

@@ -13,7 +13,7 @@
             read-timeout="PT5M"
             write-timeout="PT5M"/>
     <event-executor use-virtual-threads="true"/>
-    <cache xs:type="rbcs-memcache:memcacheCacheType" max-age="P7D" max-size="16777216" compression-mode="deflate">
+    <cache xs:type="rbcs-memcache:memcacheCacheType" max-age="P7D" max-size="16777216" compression-mode="deflate" chunk-size="123">
         <server host="memcached" port="11211"/>
     </cache>
     <authorization>

rbcs-server/src/test/resources/net/woggioni/rbcs/server/test/valid/rbcs-memcached.xml

@@ -12,7 +12,7 @@
             idle-timeout="PT30M"
             max-request-size="101325"/>
     <event-executor use-virtual-threads="false"/>
-    <cache xs:type="rbcs-memcache:memcacheCacheType" max-age="P7D" max-size="101325" digest="SHA-256">
+    <cache xs:type="rbcs-memcache:memcacheCacheType" max-age="P7D" max-size="101325" digest="SHA-256" chunk-size="456">
         <server host="127.0.0.1" port="11211" max-connections="10" connection-timeout="PT20S"/>
     </cache>
     <authentication>

rbcs-server/src/test/resources/net/woggioni/rbcs/server/test/valid/rbcs-tls.xml

@@ -11,7 +11,7 @@
             idle-timeout="PT30M"
             max-request-size="4096"/>
     <event-executor use-virtual-threads="false"/>
-    <cache xs:type="rbcs:inMemoryCacheType" max-age="P7D"/>
+    <cache xs:type="rbcs:inMemoryCacheType" max-age="P7D" chunk-size="0xa91f"/>
     <authorization>
         <users>
             <user name="user1" password="password1">

settings.gradle

@@ -29,6 +29,6 @@ include 'rbcs-api'
 include 'rbcs-common'
 include 'rbcs-server-memcache'
 include 'rbcs-cli'
-include 'docker'
 include 'rbcs-client'
 include 'rbcs-server'
+include 'docker'