Compare commits

...

16 Commits
0.1.0 ... doc

Author SHA1 Message Date
c19bc9e91e temporary commit 2025-02-07 10:18:54 +08:00
af79e74b95 fixed max message size for memcache backend 2025-02-06 23:09:22 +08:00
78ae21caa4 0.1.4 release
All checks were successful
CI / build (push) Successful in 8m14s
2025-02-06 15:24:00 +08:00
6c0eadb9fb renamed project to "Remote Cache Build Server" (RBCS) 2025-02-06 15:20:50 +08:00
5fef1b932e updated lys-catalog version
All checks were successful
CI / build (push) Successful in 2m32s
2025-02-05 21:49:08 +08:00
5e173dbf62 fixed unit tests 2025-02-05 21:24:10 +08:00
53b24e3d54 improved benchmark accuracy 2025-02-05 19:10:25 +08:00
7d0f24fa58 fixed memory leak in InMemoryCache 2025-02-05 19:09:51 +08:00
1b6cf1bd96 fixed memory leak in memcached plugin 2025-02-05 14:41:11 +08:00
4180df2352 added healthcheck command to client 2025-02-05 00:02:17 +08:00
c2e388b931 switched to ZGC in docker image
All checks were successful
CI / build (push) Successful in 3m28s
2025-02-04 22:46:34 +08:00
6c62ac85c0 implemented memcached client with Netty
All checks were successful
CI / build (push) Successful in 1m46s
2025-02-04 22:09:28 +08:00
89153b60f8 fixed race condition in InMemoryCache 2025-02-01 10:14:13 +08:00
a2a40ab60f added semaphore to benchmark command 2025-01-28 00:00:07 +08:00
45458761f3 made TLS client certificate request from the server configurable
All checks were successful
CI / build (push) Successful in 4m2s
2025-01-27 13:32:04 +08:00
90a5834f5f added retry policy to gbcs-client 2025-01-27 13:12:12 +08:00
131 changed files with 2260 additions and 1235 deletions

View File

@@ -31,7 +31,7 @@ jobs:
username: woggioni
password: ${{ secrets.PUBLISHER_TOKEN }}
-
name: Build gbcs Docker image
name: Build rbcs Docker image
uses: docker/build-push-action@v5.3.0
with:
context: "docker/build/docker"
@@ -39,12 +39,12 @@ jobs:
push: true
pull: true
tags: |
gitea.woggioni.net/woggioni/gbcs:latest
gitea.woggioni.net/woggioni/gbcs:${{ steps.retrieve-version.outputs.VERSION }}
gitea.woggioni.net/woggioni/rbcs:latest
gitea.woggioni.net/woggioni/rbcs:${{ steps.retrieve-version.outputs.VERSION }}
target: release
cache-from: type=registry,ref=gitea.woggioni.net/woggioni/gbcs:buildx
cache-from: type=registry,ref=gitea.woggioni.net/woggioni/rbcs:buildx
-
name: Build gbcs memcached Docker image
name: Build rbcs memcache Docker image
uses: docker/build-push-action@v5.3.0
with:
context: "docker/build/docker"
@@ -52,11 +52,11 @@ jobs:
push: true
pull: true
tags: |
gitea.woggioni.net/woggioni/gbcs:memcached
gitea.woggioni.net/woggioni/gbcs:memcached-${{ steps.retrieve-version.outputs.VERSION }}
target: release-memcached
cache-from: type=registry,ref=gitea.woggioni.net/woggioni/gbcs:buildx
cache-to: type=registry,mode=max,compression=zstd,image-manifest=true,oci-mediatypes=true,ref=gitea.woggioni.net/woggioni/gbcs:buildx
gitea.woggioni.net/woggioni/rbcs:memcache
gitea.woggioni.net/woggioni/rbcs:memcache-${{ steps.retrieve-version.outputs.VERSION }}
target: release-memcache
cache-from: type=registry,ref=gitea.woggioni.net/woggioni/rbcs:buildx
cache-to: type=registry,mode=max,compression=zstd,image-manifest=true,oci-mediatypes=true,ref=gitea.woggioni.net/woggioni/rbcs:buildx
- name: Publish artifacts
env:
PUBLISHER_TOKEN: ${{ secrets.PUBLISHER_TOKEN }}

2
.gitignore vendored
View File

@@ -4,4 +4,4 @@
# Ignore Gradle build output directory
build
gbcs-cli/native-image/*.json
rbcs-cli/native-image/*.json

20
LICENSE Normal file
View File

@@ -0,0 +1,20 @@
The MIT License (MIT)
Copyright (c) 2017 Y. T. CHUNG <zonyitoo@gmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

110
README.md
View File

@@ -0,0 +1,110 @@
# Remote Build Cache Server
Remote Build Cache Server (shortened to RBCS) allows you to share and reuse unchanged build
and test outputs across the team. This speeds up local and CI builds since cycles are not wasted
re-building components that are unaffected by new code changes. RBCS supports both Gradle and
Maven build tool environments.
## Getting Started
### Downloading the jar file
You can download the latest version from [this link](https://gitea.woggioni.net/woggioni/-/packages/maven/net.woggioni-rbcs-cli/)
If you want to use memcache as a storage backend you'll also need to download [the memcache plugin](https://gitea.woggioni.net/woggioni/-/packages/maven/net.woggioni-rbcs-server-memcache/)
### Using the Docker image
You can pull the latest Docker image with
```bash
docker pull gitea.woggioni.net/woggioni/rbcs:latest
```
## Usage
## Configuration
### Using RBCS with Gradle
```groovy
buildCache {
remote(HttpBuildCache) {
url = 'https://rbcs.example.com/'
}
}
```
### Using RBCS with Maven
Read [here](https://maven.apache.org/extensions/maven-build-cache-extension/remote-cache.html)
## FAQ
### Why should I use a build cache?
#### Build Caches Improve Build & Test Performance
Building software consists of a number of steps, like compiling sources, executing tests, and linking binaries. Weve seen that a binary artifact repository helps when such a step requires an external component by downloading the artifact from the repository rather than building it locally.
However, there are many additional steps in this build process which can be optimized to reduce the build time. An obvious strategy is to avoid executing build steps which dominate the total build time when these build steps are not needed.
Most build times are dominated by the testing step.
While binary repositories cannot capture the outcome of a test build step (only the test reports
when included in binary artifacts), build caches are designed to eliminate redundant executions
for every build step. Moreover, it generalizes the concept of avoiding work associated with any
incremental step of the build, including test execution, compilation and resource processing.
The mechanism itself is comparable to a pure function. That is, given some inputs such as source
files and environment parameters we know that the output is always going to be the same.
As a result, we can cache it and retrieve it based on a simple cryptographic hash of the inputs.
Build caching is supported natively by some build tools.
#### Improve CI builds with a remote build cache
When analyzing the role of a build cache it is important to take into account the granularity
of the changes that it caches. Imagine a full build for a project with 40 to 50 modules
which fails at the last step (deployment) because the staging environment is temporarily unavailable.
Although the vast majority of the build steps (potentially thousands) succeed,
the change can not be deployed to the staging environment.
Without a build cache one typically relies on a very complex CI configuration to reuse build step outputs
or would have to repeat the full build once the environment is available.
Some build tools dont support incremental builds properly. For example, outputs of a build started
from scratch may vary when compared to subsequent builds that rely on the initial builds output.
As a result, to preserve build integrity, its crucial to rebuild from scratch, or cleanly, in this
scenario.
With a build cache, only the last step needs to be executed and the build can be re-triggered
when the environment is back online. This automatically saves all of the time and
resources required across the different build steps which were successfully executed.
Instead of executing the intermediate steps, the build tool pulls the outputs from the build cache,
avoiding a lot of redundant work
#### Share outputs with a remote build cache
One of the most important advantages of a remote build cache is the ability to share build outputs.
In most CI configurations, for example, a number of pipelines are created.
These may include one for building the sources, one for testing, one for publishing the outcomes
to a remote repository, and other pipelines to test on different platforms.
There are even situations where CI builds partially build a project (i.e. some modules and not others).
Most of those pipelines share a lot of intermediate build steps. All builds which perform testing
require the binaries to be ready. All publishing builds require all previous steps to be executed.
And because modern CI infrastructure means executing everything in containerized (isolated) environments,
significant resources are wasted by repeatedly building the same intermediate artifacts.
A remote build cache greatly reduces this overhead by orders of magnitudes because it provides a way
for all those pipelines to share their outputs. After all, there is no point recreating an output that
is already available in the cache.
Because there are inherent dependencies between software components of a build,
introducing a build cache dramatically reduces the impact of exploding a component into multiple pieces,
allowing for increased modularity without increased overhead.
#### Make local developers more efficient with remote build caches
It is common for different teams within a company to work on different modules of a single large
application. In this case, most teams dont care about building the other parts of the software.
By introducing a remote cache developers immediately benefit from pre-built artifacts when checking out code.
Because it has already been built on CI, they dont have to do it locally.
Introducing a remote cache is a huge benefit for those developers. Consider that a typical developers
day begins by performing a code checkout. Most likely the checked out code has already been built on CI.
Therefore, no time is wasted running the first build of the day. The remote cache provides all of the
intermediate artifacts needed. And, in the event local changes are made, the remote cache still leverages
partial cache hits for projects which are independent. As other developers in the organization request
CI builds, the remote cache continues to populate, increasing the likelihood of these remote cache hits
across team members.

View File

@@ -15,7 +15,7 @@ allprojects { subproject ->
version = project.currentTag.map { it[0] }.get()
} else {
version = project.gitRevision.map { gitRevision ->
"${getProperty('gbcs.version')}.${gitRevision[0..10]}"
"${getProperty('rbcs.version')}.${gitRevision[0..10]}"
}.get()
}
@@ -46,6 +46,12 @@ allprojects { subproject ->
}
}
dependencies {
testImplementation catalog.junit.jupiter.api
testImplementation catalog.junit.jupiter.params
testRuntimeOnly catalog.junit.jupiter.engine
}
test {
useJUnitPlatform()
}

View File

@@ -4,13 +4,13 @@ USER luser
WORKDIR /home/luser
FROM base-release AS release
ADD gbcs-cli-envelope-*.jar gbcs.jar
ENTRYPOINT ["java", "-jar", "/home/luser/gbcs.jar", "server"]
ADD rbcs-cli-envelope-*.jar rbcs.jar
ENTRYPOINT ["java", "-XX:+UseZGC", "-XX:+ZGenerational", "-jar", "/home/luser/rbcs.jar", "server"]
FROM base-release AS release-memcached
ADD --chown=luser:luser gbcs-cli-envelope-*.jar gbcs.jar
FROM base-release AS release-memcache
ADD --chown=luser:luser rbcs-cli-envelope-*.jar rbcs.jar
RUN mkdir plugins
WORKDIR /home/luser/plugins
RUN --mount=type=bind,source=.,target=/build/distributions tar -xf /build/distributions/gbcs-server-memcached*.tar
RUN --mount=type=bind,source=.,target=/build/distributions tar -xf /build/distributions/rbcs-server-memcache*.tar
WORKDIR /home/luser
ENTRYPOINT ["java", "-jar", "/home/luser/gbcs.jar", "server"]
ENTRYPOINT ["java", "-XX:+UseZGC", "-XX:+ZGenerational", "-jar", "/home/luser/rbcs.jar", "server"]

View File

@@ -18,8 +18,8 @@ configurations {
}
dependencies {
docker project(path: ':gbcs-cli', configuration: 'release')
docker project(path: ':gbcs-server-memcached', configuration: 'release')
docker project(path: ':rbcs-cli', configuration: 'release')
docker project(path: ':rbcs-server-memcache', configuration: 'release')
}
Provider<Task> cleanTaskProvider = tasks.named(BasePlugin.CLEAN_TASK_NAME) {}
@@ -35,33 +35,33 @@ Provider<Copy> prepareDockerBuild = tasks.register('prepareDockerBuild', Copy) {
Provider<DockerBuildImage> dockerBuild = tasks.register('dockerBuildImage', DockerBuildImage) {
group = 'docker'
dependsOn prepareDockerBuild
images.add('gitea.woggioni.net/woggioni/gbcs:latest')
images.add("gitea.woggioni.net/woggioni/gbcs:${version}")
images.add('gitea.woggioni.net/woggioni/rbcs:latest')
images.add("gitea.woggioni.net/woggioni/rbcs:${version}")
}
Provider<DockerTagImage> dockerTag = tasks.register('dockerTagImage', DockerTagImage) {
group = 'docker'
repository = 'gitea.woggioni.net/woggioni/gbcs'
imageId = 'gitea.woggioni.net/woggioni/gbcs:latest'
repository = 'gitea.woggioni.net/woggioni/rbcs'
imageId = 'gitea.woggioni.net/woggioni/rbcs:latest'
tag = version
}
Provider<DockerTagImage> dockerTagMemcached = tasks.register('dockerTagMemcachedImage', DockerTagImage) {
Provider<DockerTagImage> dockerTagMemcache = tasks.register('dockerTagMemcacheImage', DockerTagImage) {
group = 'docker'
repository = 'gitea.woggioni.net/woggioni/gbcs'
imageId = 'gitea.woggioni.net/woggioni/gbcs:memcached'
tag = "${version}-memcached"
repository = 'gitea.woggioni.net/woggioni/rbcs'
imageId = 'gitea.woggioni.net/woggioni/rbcs:memcache'
tag = "${version}-memcache"
}
Provider<DockerPushImage> dockerPush = tasks.register('dockerPushImage', DockerPushImage) {
group = 'docker'
dependsOn dockerTag, dockerTagMemcached
dependsOn dockerTag, dockerTagMemcache
registryCredentials {
url = getProperty('docker.registry.url')
username = 'woggioni'
password = System.getenv().get("PUBLISHER_TOKEN")
}
images = [dockerTag.flatMap{ it.tag }, dockerTagMemcached.flatMap{ it.tag }]
images = [dockerTag.flatMap{ it.tag }, dockerTagMemcache.flatMap{ it.tag }]
}

View File

@@ -1,6 +0,0 @@
module net.woggioni.gbcs.api {
requires static lombok;
requires java.xml;
exports net.woggioni.gbcs.api;
exports net.woggioni.gbcs.api.exception;
}

View File

@@ -1,12 +0,0 @@
package net.woggioni.gbcs.api;
import net.woggioni.gbcs.api.exception.ContentTooLargeException;
import java.nio.channels.ReadableByteChannel;
public interface Cache extends AutoCloseable {
ReadableByteChannel get(String key);
void put(String key, byte[] content) throws ContentTooLargeException;
}

View File

@@ -1,7 +0,0 @@
package net.woggioni.gbcs.api.exception;
public class GbcsException extends RuntimeException {
public GbcsException(String message, Throwable cause) {
super(message, cause);
}
}

View File

@@ -1,17 +0,0 @@
module net.woggioni.gbcs.cli {
requires org.slf4j;
requires net.woggioni.gbcs.server;
requires info.picocli;
requires net.woggioni.gbcs.common;
requires net.woggioni.gbcs.client;
requires kotlin.stdlib;
requires net.woggioni.jwo;
requires net.woggioni.gbcs.api;
exports net.woggioni.gbcs.cli.impl.converters to info.picocli;
opens net.woggioni.gbcs.cli.impl.commands to info.picocli;
opens net.woggioni.gbcs.cli.impl to info.picocli;
opens net.woggioni.gbcs.cli to info.picocli, net.woggioni.gbcs.common;
exports net.woggioni.gbcs.cli;
}

View File

@@ -1,64 +0,0 @@
package net.woggioni.gbcs.cli
import net.woggioni.gbcs.common.GbcsUrlStreamHandlerFactory
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.cli.impl.AbstractVersionProvider
import net.woggioni.gbcs.cli.impl.GbcsCommand
import net.woggioni.gbcs.cli.impl.commands.BenchmarkCommand
import net.woggioni.gbcs.cli.impl.commands.ClientCommand
import net.woggioni.gbcs.cli.impl.commands.GetCommand
import net.woggioni.gbcs.cli.impl.commands.PasswordHashCommand
import net.woggioni.gbcs.cli.impl.commands.PutCommand
import net.woggioni.gbcs.cli.impl.commands.ServerCommand
import net.woggioni.jwo.Application
import picocli.CommandLine
import picocli.CommandLine.Model.CommandSpec
import java.net.URI
@CommandLine.Command(
name = "gbcs", versionProvider = GradleBuildCacheServerCli.VersionProvider::class
)
class GradleBuildCacheServerCli : GbcsCommand() {
class VersionProvider : AbstractVersionProvider()
companion object {
@JvmStatic
fun main(vararg args: String) {
Thread.currentThread().contextClassLoader = GradleBuildCacheServerCli::class.java.classLoader
GbcsUrlStreamHandlerFactory.install()
val log = contextLogger()
val app = Application.builder("gbcs")
.configurationDirectoryEnvVar("GBCS_CONFIGURATION_DIR")
.configurationDirectoryPropertyKey("net.woggioni.gbcs.conf.dir")
.build()
val gbcsCli = GradleBuildCacheServerCli()
val commandLine = CommandLine(gbcsCli)
commandLine.setExecutionExceptionHandler { ex, cl, parseResult ->
log.error(ex.message, ex)
CommandLine.ExitCode.SOFTWARE
}
commandLine.addSubcommand(ServerCommand(app))
commandLine.addSubcommand(PasswordHashCommand())
commandLine.addSubcommand(
CommandLine(ClientCommand(app)).apply {
addSubcommand(BenchmarkCommand())
addSubcommand(PutCommand())
addSubcommand(GetCommand())
})
System.exit(commandLine.execute(*args))
}
}
@CommandLine.Option(names = ["-V", "--version"], versionHelp = true)
var versionHelp = false
private set
@CommandLine.Spec
private lateinit var spec: CommandSpec
override fun run() {
spec.commandLine().usage(System.out);
}
}

View File

@@ -1,134 +0,0 @@
package net.woggioni.gbcs.cli.impl.commands
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.common.error
import net.woggioni.gbcs.common.info
import net.woggioni.gbcs.cli.impl.GbcsCommand
import net.woggioni.gbcs.client.GradleBuildCacheClient
import net.woggioni.jwo.JWO
import picocli.CommandLine
import java.security.SecureRandom
import java.time.Duration
import java.time.Instant
import java.util.Base64
import java.util.concurrent.ExecutionException
import java.util.concurrent.Future
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.atomic.AtomicLong
import kotlin.random.Random
@CommandLine.Command(
name = "benchmark",
description = ["Run a load test against the server"],
showDefaultValues = true
)
class BenchmarkCommand : GbcsCommand() {
private val log = contextLogger()
@CommandLine.Spec
private lateinit var spec: CommandLine.Model.CommandSpec
@CommandLine.Option(
names = ["-e", "--entries"],
description = ["Total number of elements to be added to the cache"],
paramLabel = "NUMBER_OF_ENTRIES"
)
private var numberOfEntries = 1000
override fun run() {
val clientCommand = spec.parent().userObject() as ClientCommand
val profile = clientCommand.profileName.let { profileName ->
clientCommand.configuration.profiles[profileName]
?: throw IllegalArgumentException("Profile $profileName does not exist in configuration")
}
val client = GradleBuildCacheClient(profile)
val entryGenerator = sequence {
val random = Random(SecureRandom.getInstance("NativePRNGNonBlocking").nextLong())
while (true) {
val key = JWO.bytesToHex(random.nextBytes(16))
val content = random.nextInt().toByte()
val value = ByteArray(0x1000, { _ -> content })
yield(key to value)
}
}
val entries = let {
val completionQueue = LinkedBlockingQueue<Future<Pair<String, ByteArray>>>(numberOfEntries)
val start = Instant.now()
val totalElapsedTime = AtomicLong(0)
entryGenerator.take(numberOfEntries).forEach { entry ->
val requestStart = System.nanoTime()
val future = client.put(entry.first, entry.second).thenApply { entry }
future.whenComplete { _, _ ->
totalElapsedTime.addAndGet((System.nanoTime() - requestStart))
completionQueue.put(future)
}
}
val inserted = sequence<Pair<String, ByteArray>> {
var completionCounter = 0
while (completionCounter < numberOfEntries) {
val future = completionQueue.take()
try {
yield(future.get())
} catch (ee: ExecutionException) {
val cause = ee.cause ?: ee
log.error(cause.message, cause)
}
completionCounter += 1
}
}.toList()
val end = Instant.now()
log.info {
val elapsed = Duration.between(start, end).toMillis()
"Insertion rate: ${numberOfEntries.toDouble() / elapsed * 1000} ops/s"
}
log.info {
"Average time per insertion: ${totalElapsedTime.get() / numberOfEntries.toDouble() * 1000} ms"
}
inserted
}
log.info {
"Inserted ${entries.size} entries"
}
if (entries.isNotEmpty()) {
val completionQueue = LinkedBlockingQueue<Future<Unit>>(entries.size)
val start = Instant.now()
val totalElapsedTime = AtomicLong(0)
entries.forEach { entry ->
val requestStart = System.nanoTime()
val future = client.get(entry.first).thenApply {
totalElapsedTime.addAndGet((System.nanoTime() - requestStart))
if (it == null) {
log.error {
"Missing entry for key '${entry.first}'"
}
} else if (!entry.second.contentEquals(it)) {
log.error {
"Retrieved a value different from what was inserted for key '${entry.first}'"
}
}
}
future.whenComplete { _, _ ->
completionQueue.put(future)
}
}
var completionCounter = 0
while (completionCounter < entries.size) {
completionQueue.take()
completionCounter += 1
}
val end = Instant.now()
log.info {
val elapsed = Duration.between(start, end).toMillis()
"Retrieval rate: ${entries.size.toDouble() / elapsed * 1000} ops/s"
}
log.info {
"Average time per retrieval: ${totalElapsedTime.get() / numberOfEntries.toDouble() * 1e6} ms"
}
} else {
log.error("Skipping retrieval benchmark as it was not possible to insert any entry in the cache")
}
}
}

View File

@@ -1,72 +0,0 @@
package net.woggioni.gbcs.client.impl
import net.woggioni.gbcs.api.exception.ConfigurationException
import net.woggioni.gbcs.common.Xml.Companion.asIterable
import net.woggioni.gbcs.common.Xml.Companion.renderAttribute
import net.woggioni.gbcs.client.GradleBuildCacheClient
import org.w3c.dom.Document
import java.net.URI
import java.nio.file.Files
import java.nio.file.Path
import java.security.KeyStore
import java.security.PrivateKey
import java.security.cert.X509Certificate
import java.time.Duration
object Parser {
fun parse(document: Document): GradleBuildCacheClient.Configuration {
val root = document.documentElement
val profiles = mutableMapOf<String, GradleBuildCacheClient.Configuration.Profile>()
for (child in root.asIterable()) {
val tagName = child.localName
when (tagName) {
"profile" -> {
val name = child.renderAttribute("name") ?: throw ConfigurationException("name attribute is required")
val uri = child.renderAttribute("base-url")?.let(::URI) ?: throw ConfigurationException("base-url attribute is required")
var authentication: GradleBuildCacheClient.Configuration.Authentication? = null
for (gchild in child.asIterable()) {
when (gchild.localName) {
"tls-client-auth" -> {
val keyStoreFile = gchild.renderAttribute("key-store-file")
val keyStorePassword =
gchild.renderAttribute("key-store-password")
val keyAlias = gchild.renderAttribute("key-alias")
val keyPassword = gchild.renderAttribute("key-password")
val keystore = KeyStore.getInstance("PKCS12").apply {
Files.newInputStream(Path.of(keyStoreFile)).use {
load(it, keyStorePassword?.toCharArray())
}
}
val key = keystore.getKey(keyAlias, keyPassword?.toCharArray()) as PrivateKey
val certChain = keystore.getCertificateChain(keyAlias).asSequence()
.map { it as X509Certificate }
.toList()
.toTypedArray()
authentication =
GradleBuildCacheClient.Configuration.Authentication.TlsClientAuthenticationCredentials(key, certChain)
}
"basic-auth" -> {
val username = gchild.renderAttribute("user") ?: throw ConfigurationException("username attribute is required")
val password = gchild.renderAttribute("password") ?: throw ConfigurationException("password attribute is required")
authentication =
GradleBuildCacheClient.Configuration.Authentication.BasicAuthenticationCredentials(username, password)
}
}
}
val maxConnections = child.renderAttribute("max-connections")
?.let(String::toInt)
?: 50
val connectionTimeout = child.renderAttribute("connection-timeout")
?.let(Duration::parse)
profiles[name] = GradleBuildCacheClient.Configuration.Profile(uri, authentication, connectionTimeout, maxConnections)
}
}
}
return GradleBuildCacheClient.Configuration(profiles)
}
}

View File

@@ -1,16 +0,0 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<gbcs-client:profiles xmlns:xs="http://www.w3.org/2001/XMLSchema-instance"
xmlns:gbcs-client="urn:net.woggioni.gbcs.client"
xs:schemaLocation="urn:net.woggioni.gbcs.client jms://net.woggioni.gbcs.client/net/woggioni/gbcs/client/schema/gbcs-client.xsd"
>
<profile name="profile1" base-url="https://gbcs1.example.com/">
<tls-client-auth
key-store-file="keystore.pfx"
key-store-password="password"
key-alias="woggioni@c962475fa38"
key-password="key-password"/>
</profile>
<profile name="profile2" base-url="https://gbcs2.example.com/">
<basic-auth user="user" password="password"/>
</profile>
</gbcs-client:profiles>

View File

@@ -1,12 +0,0 @@
package net.woggioni.gbcs.common
import java.net.URI
import java.net.URL
object GBCS {
fun String.toUrl() : URL = URL.of(URI(this), null)
const val GBCS_NAMESPACE_URI: String = "urn:net.woggioni.gbcs.server"
const val GBCS_PREFIX: String = "gbcs"
const val XML_SCHEMA_NAMESPACE_URI = "http://www.w3.org/2001/XMLSchema-instance"
}

View File

@@ -1 +0,0 @@
net.woggioni.gbcs.common.GbcsUrlStreamHandlerFactory

View File

@@ -1,14 +0,0 @@
import net.woggioni.gbcs.api.CacheProvider;
module net.woggioni.gbcs.server.memcached {
requires net.woggioni.gbcs.common;
requires net.woggioni.gbcs.api;
requires com.googlecode.xmemcached;
requires net.woggioni.jwo;
requires java.xml;
requires kotlin.stdlib;
provides CacheProvider with net.woggioni.gbcs.server.memcached.MemcachedCacheProvider;
opens net.woggioni.gbcs.server.memcached.schema;
}

View File

@@ -1,59 +0,0 @@
package net.woggioni.gbcs.server.memcached
import net.rubyeye.xmemcached.XMemcachedClientBuilder
import net.rubyeye.xmemcached.command.BinaryCommandFactory
import net.rubyeye.xmemcached.transcoders.CompressionMode
import net.rubyeye.xmemcached.transcoders.SerializingTranscoder
import net.woggioni.gbcs.api.Cache
import net.woggioni.gbcs.api.exception.ContentTooLargeException
import net.woggioni.gbcs.common.HostAndPort
import net.woggioni.jwo.JWO
import java.io.ByteArrayInputStream
import java.net.InetSocketAddress
import java.nio.channels.Channels
import java.nio.channels.ReadableByteChannel
import java.nio.charset.StandardCharsets
import java.security.MessageDigest
import java.time.Duration
class MemcachedCache(
servers: List<HostAndPort>,
private val maxAge: Duration,
maxSize : Int,
digestAlgorithm: String?,
compressionMode: CompressionMode,
) : Cache {
private val memcachedClient = XMemcachedClientBuilder(
servers.stream().map { addr: HostAndPort -> InetSocketAddress(addr.host, addr.port) }.toList()
).apply {
commandFactory = BinaryCommandFactory()
digestAlgorithm?.let { dAlg ->
setKeyProvider { key ->
val md = MessageDigest.getInstance(dAlg)
md.update(key.toByteArray(StandardCharsets.UTF_8))
JWO.bytesToHex(md.digest())
}
}
transcoder = SerializingTranscoder(maxSize).apply {
setCompressionMode(compressionMode)
}
}.build()
override fun get(key: String): ReadableByteChannel? {
return memcachedClient.get<ByteArray>(key)
?.let(::ByteArrayInputStream)
?.let(Channels::newChannel)
}
override fun put(key: String, content: ByteArray) {
try {
memcachedClient[key, maxAge.toSeconds().toInt()] = content
} catch (e: IllegalArgumentException) {
throw ContentTooLargeException(e.message, e)
}
}
override fun close() {
memcachedClient.shutdown()
}
}

View File

@@ -1,26 +0,0 @@
package net.woggioni.gbcs.server.memcached
import net.rubyeye.xmemcached.transcoders.CompressionMode
import net.woggioni.gbcs.api.Configuration
import net.woggioni.gbcs.common.HostAndPort
import java.time.Duration
data class MemcachedCacheConfiguration(
var servers: List<HostAndPort>,
var maxAge: Duration = Duration.ofDays(1),
var maxSize: Int = 0x100000,
var digestAlgorithm: String? = null,
var compressionMode: CompressionMode = CompressionMode.ZIP,
) : Configuration.Cache {
override fun materialize() = MemcachedCache(
servers,
maxAge,
maxSize,
digestAlgorithm,
compressionMode
)
override fun getNamespaceURI() = "urn:net.woggioni.gbcs.server.memcached"
override fun getTypeName() = "memcachedCacheType"
}

View File

@@ -1,88 +0,0 @@
package net.woggioni.gbcs.server.memcached
import net.rubyeye.xmemcached.transcoders.CompressionMode
import net.woggioni.gbcs.api.CacheProvider
import net.woggioni.gbcs.api.exception.ConfigurationException
import net.woggioni.gbcs.common.GBCS
import net.woggioni.gbcs.common.HostAndPort
import net.woggioni.gbcs.common.Xml
import net.woggioni.gbcs.common.Xml.Companion.asIterable
import net.woggioni.gbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import org.w3c.dom.Element
import java.time.Duration
class MemcachedCacheProvider : CacheProvider<MemcachedCacheConfiguration> {
override fun getXmlSchemaLocation() = "jpms://net.woggioni.gbcs.server.memcached/net/woggioni/gbcs/server/memcached/schema/gbcs-memcached.xsd"
override fun getXmlType() = "memcachedCacheType"
override fun getXmlNamespace() = "urn:net.woggioni.gbcs.server.memcached"
val xmlNamespacePrefix : String
get() = "gbcs-memcached"
override fun deserialize(el: Element): MemcachedCacheConfiguration {
val servers = mutableListOf<HostAndPort>()
val maxAge = el.renderAttribute("max-age")
?.let(Duration::parse)
?: Duration.ofDays(1)
val maxSize = el.renderAttribute("max-size")
?.let(String::toInt)
?: 0x100000
val compressionMode = el.renderAttribute("compression-mode")
?.let {
when (it) {
"gzip" -> CompressionMode.GZIP
"zip" -> CompressionMode.ZIP
else -> CompressionMode.ZIP
}
}
?: CompressionMode.ZIP
val digestAlgorithm = el.renderAttribute("digest")
for (child in el.asIterable()) {
when (child.nodeName) {
"server" -> {
val host = child.renderAttribute("host") ?: throw ConfigurationException("host attribute is required")
val port = child.renderAttribute("port")?.toInt() ?: throw ConfigurationException("port attribute is required")
servers.add(HostAndPort(host, port))
}
}
}
return MemcachedCacheConfiguration(
servers,
maxAge,
maxSize,
digestAlgorithm,
compressionMode,
)
}
override fun serialize(doc: Document, cache: MemcachedCacheConfiguration) = cache.run {
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", GBCS.XML_SCHEMA_NAMESPACE_URI)
for (server in servers) {
node("server") {
attr("host", server.host)
attr("port", server.port.toString())
}
}
attr("max-age", maxAge.toString())
attr("max-size", maxSize.toString())
digestAlgorithm?.let { digestAlgorithm ->
attr("digest", digestAlgorithm)
}
attr(
"compression-mode", when (compressionMode) {
CompressionMode.GZIP -> "gzip"
CompressionMode.ZIP -> "zip"
}
)
}
result
}
}

View File

@@ -1 +0,0 @@
net.woggioni.gbcs.server.memcached.MemcachedCacheProvider

View File

@@ -1,106 +0,0 @@
package net.woggioni.gbcs.server.cache
import net.woggioni.gbcs.api.Cache
import net.woggioni.gbcs.server.cache.CacheUtils.digestString
import java.io.ByteArrayInputStream
import java.io.ByteArrayOutputStream
import java.nio.ByteBuffer
import java.nio.channels.Channels
import java.security.MessageDigest
import java.time.Duration
import java.time.Instant
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.PriorityBlockingQueue
import java.util.concurrent.atomic.AtomicInteger
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 InMemoryCache(
val maxAge: Duration,
val digestAlgorithm: String?,
val compressionEnabled: Boolean,
val compressionLevel: Int
) : Cache {
private val map = ConcurrentHashMap<String, MapValue>()
private class MapValue(val rc: AtomicInteger, val payload : AtomicReference<ByteArray>)
private class RemovalQueueElement(val key: String, val expiry : Instant) : Comparable<RemovalQueueElement> {
override fun compareTo(other: RemovalQueueElement)= expiry.compareTo(other.expiry)
}
private val removalQueue = PriorityBlockingQueue<RemovalQueueElement>()
private var running = true
private val garbageCollector = Thread({
while(true) {
val el = removalQueue.take()
val now = Instant.now()
if(now > el.expiry) {
val value = map[el.key] ?: continue
val rc = value.rc.decrementAndGet()
if(rc == 0) {
map.remove(el.key)
}
} else {
removalQueue.put(el)
Thread.sleep(minOf(Duration.between(now, el.expiry), Duration.ofSeconds(1)))
}
}
}).apply {
start()
}
override fun close() {
running = false
garbageCollector.join()
}
override fun get(key: String) =
(digestAlgorithm
?.let(MessageDigest::getInstance)
?.let { md ->
digestString(key.toByteArray(), md)
} ?: key
).let { digest ->
map[digest]
?.let(MapValue::payload)
?.let(AtomicReference<ByteArray>::get)
?.let { value ->
if (compressionEnabled) {
val inflater = Inflater()
Channels.newChannel(InflaterInputStream(ByteArrayInputStream(value), inflater))
} else {
Channels.newChannel(ByteArrayInputStream(value))
}
}
}
override fun put(key: String, content: ByteArray) {
(digestAlgorithm
?.let(MessageDigest::getInstance)
?.let { md ->
digestString(key.toByteArray(), md)
} ?: key).let { digest ->
val value = if (compressionEnabled) {
val deflater = Deflater(compressionLevel)
val baos = ByteArrayOutputStream()
DeflaterOutputStream(baos, deflater).use { stream ->
stream.write(content)
}
baos.toByteArray()
} else {
content
}
val mapValue = map.computeIfAbsent(digest) {
MapValue(AtomicInteger(0), AtomicReference())
}
mapValue.payload.set(value)
removalQueue.put(RemovalQueueElement(digest, Instant.now().plus(maxAge)))
}
}
}

View File

@@ -1,2 +0,0 @@
net.woggioni.gbcs.server.cache.FileSystemCacheProvider
net.woggioni.gbcs.server.cache.InMemoryCacheProvider

View File

@@ -1,21 +0,0 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<gbcs:server xmlns:xs="http://www.w3.org/2001/XMLSchema-instance"
xmlns:gbcs="urn:net.woggioni.gbcs.server"
xmlns:gbcs-memcached="urn:net.woggioni.gbcs.server.memcached"
xs:schemaLocation="urn:net.woggioni.gbcs.server.memcached jpms://net.woggioni.gbcs.server.memcached/net/woggioni/gbcs/server/memcached/schema/gbcs-memcached.xsd urn:net.woggioni.gbcs.server jpms://net.woggioni.gbcs.server/net/woggioni/gbcs/server/schema/gbcs.xsd">
<bind host="127.0.0.1" port="11443" incoming-connections-backlog-size="50"/>
<connection
write-timeout="PT25M"
read-timeout="PT20M"
read-idle-timeout="PT10M"
write-idle-timeout="PT11M"
idle-timeout="PT30M"
max-request-size="101325"/>
<event-executor use-virtual-threads="false"/>
<cache xs:type="gbcs-memcached:memcachedCacheType" max-age="P7D" max-size="101325" digest="SHA-256">
<server host="127.0.0.1" port="11211"/>
</cache>
<authentication>
<none/>
</authentication>
</gbcs:server>

View File

@@ -2,9 +2,9 @@ org.gradle.configuration-cache=false
org.gradle.parallel=true
org.gradle.caching=true
gbcs.version = 0.0.11
rbcs.version = 0.1.4
lys.version = 2025.01.25
lys.version = 2025.02.05
gitea.maven.url = https://gitea.woggioni.net/api/packages/woggioni/maven
docker.registry.url=gitea.woggioni.net

View File

@@ -5,6 +5,7 @@ plugins {
}
dependencies {
api catalog.netty.buffer
}
publishing {

View File

@@ -0,0 +1,7 @@
module net.woggioni.rbcs.api {
requires static lombok;
requires java.xml;
requires io.netty.buffer;
exports net.woggioni.rbcs.api;
exports net.woggioni.rbcs.api.exception;
}

View File

@@ -0,0 +1,14 @@
package net.woggioni.rbcs.api;
import io.netty.buffer.ByteBuf;
import net.woggioni.rbcs.api.exception.ContentTooLargeException;
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;
}

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.api;
package net.woggioni.rbcs.api;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.api;
package net.woggioni.rbcs.api;
import lombok.EqualsAndHashCode;
@@ -56,7 +56,8 @@ public class Configuration {
@EqualsAndHashCode.Include
String name;
Set<Role> roles;
Quota quota;
Quota groupQuota;
Quota userQuota;
}
@Value
@@ -91,11 +92,14 @@ public class Configuration {
boolean verifyClients;
}
public enum ClientCertificate {
REQUIRED, OPTIONAL
}
@Value
public static class Tls {
KeyStore keyStore;
TrustStore trustStore;
boolean verifyClients;
}
@Value
@@ -111,6 +115,7 @@ public class Configuration {
Path file;
String password;
boolean checkCertificateStatus;
boolean requireClientCertificate;
}
@Value
@@ -130,7 +135,7 @@ public class Configuration {
}
public interface Cache {
net.woggioni.gbcs.api.Cache materialize();
net.woggioni.rbcs.api.Cache materialize();
String getNamespaceURI();
String getTypeName();
}

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.api;
package net.woggioni.rbcs.api;
public enum Role {
Reader, Writer

View File

@@ -1,6 +1,6 @@
package net.woggioni.gbcs.api.exception;
package net.woggioni.rbcs.api.exception;
public class CacheException extends GbcsException {
public class CacheException extends RbcsException {
public CacheException(String message, Throwable cause) {
super(message, cause);
}

View File

@@ -1,6 +1,6 @@
package net.woggioni.gbcs.api.exception;
package net.woggioni.rbcs.api.exception;
public class ConfigurationException extends GbcsException {
public class ConfigurationException extends RbcsException {
public ConfigurationException(String message, Throwable cause) {
super(message, cause);
}

View File

@@ -1,6 +1,6 @@
package net.woggioni.gbcs.api.exception;
package net.woggioni.rbcs.api.exception;
public class ContentTooLargeException extends GbcsException {
public class ContentTooLargeException extends RbcsException {
public ContentTooLargeException(String message, Throwable cause) {
super(message, cause);
}

View File

@@ -0,0 +1,7 @@
package net.woggioni.rbcs.api.exception;
public class RbcsException extends RuntimeException {
public RbcsException(String message, Throwable cause) {
super(message, cause);
}
}

View File

@@ -17,9 +17,9 @@ import net.woggioni.gradle.graalvm.JlinkPlugin
import net.woggioni.gradle.graalvm.JlinkTask
Property<String> mainModuleName = objects.property(String.class)
mainModuleName.set('net.woggioni.gbcs.cli')
mainModuleName.set('net.woggioni.rbcs.cli')
Property<String> mainClassName = objects.property(String.class)
mainClassName.set('net.woggioni.gbcs.cli.GradleBuildCacheServerCli')
mainClassName.set('net.woggioni.rbcs.cli.RemoteBuildCacheServerCli')
tasks.named(JavaPlugin.COMPILE_JAVA_TASK_NAME, JavaCompile) {
options.javaModuleMainClass = mainClassName
@@ -47,8 +47,8 @@ dependencies {
implementation catalog.netty.codec.http
implementation catalog.picocli
implementation project(':gbcs-client')
implementation project(':gbcs-server')
implementation project(':rbcs-client')
implementation project(':rbcs-server')
// runtimeOnly catalog.slf4j.jdk14
runtimeOnly catalog.logback.classic
@@ -56,10 +56,10 @@ dependencies {
}
Provider<EnvelopeJarTask> envelopeJarTaskProvider = tasks.named('envelopeJar', EnvelopeJarTask.class) {
// systemProperties['java.util.logging.config.class'] = 'net.woggioni.gbcs.LoggingConfig'
// systemProperties['log.config.source'] = 'net/woggioni/gbcs/cli/logging.properties'
// systemProperties['java.util.logging.config.file'] = 'classpath:net/woggioni/gbcs/cli/logging.properties'
systemProperties['logback.configurationFile'] = 'classpath:net/woggioni/gbcs/cli/logback.xml'
// systemProperties['java.util.logging.config.class'] = 'net.woggioni.rbcs.LoggingConfig'
// systemProperties['log.config.source'] = 'net/woggioni/rbcs/cli/logging.properties'
// systemProperties['java.util.logging.config.file'] = 'classpath:net/woggioni/rbcs/cli/logging.properties'
systemProperties['logback.configurationFile'] = 'classpath:net/woggioni/rbcs/cli/logback.xml'
systemProperties['io.netty.leakDetectionLevel'] = 'DISABLED'
// systemProperties['org.slf4j.simpleLogger.showDateTime'] = 'true'
@@ -83,7 +83,7 @@ tasks.named(NativeImagePlugin.NATIVE_IMAGE_TASK_NAME, NativeImageTask) {
tasks.named(JlinkPlugin.JLINK_TASK_NAME, JlinkTask) {
mainClass = mainClassName
mainModule = 'net.woggioni.gbcs.cli'
mainModule = 'net.woggioni.rbcs.cli'
}
artifacts {

View File

@@ -0,0 +1,17 @@
module net.woggioni.rbcs.cli {
requires org.slf4j;
requires net.woggioni.rbcs.server;
requires info.picocli;
requires net.woggioni.rbcs.common;
requires net.woggioni.rbcs.client;
requires kotlin.stdlib;
requires net.woggioni.jwo;
requires net.woggioni.rbcs.api;
exports net.woggioni.rbcs.cli.impl.converters to info.picocli;
opens net.woggioni.rbcs.cli.impl.commands to info.picocli;
opens net.woggioni.rbcs.cli.impl to info.picocli;
opens net.woggioni.rbcs.cli to info.picocli, net.woggioni.rbcs.common;
exports net.woggioni.rbcs.cli;
}

View File

@@ -0,0 +1,69 @@
package net.woggioni.rbcs.cli
import net.woggioni.rbcs.cli.impl.AbstractVersionProvider
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.cli.impl.commands.BenchmarkCommand
import net.woggioni.rbcs.cli.impl.commands.ClientCommand
import net.woggioni.rbcs.cli.impl.commands.GetCommand
import net.woggioni.rbcs.cli.impl.commands.HealthCheckCommand
import net.woggioni.rbcs.cli.impl.commands.PasswordHashCommand
import net.woggioni.rbcs.cli.impl.commands.PutCommand
import net.woggioni.rbcs.cli.impl.commands.ServerCommand
import net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory
import net.woggioni.rbcs.common.contextLogger
import net.woggioni.jwo.Application
import picocli.CommandLine
import picocli.CommandLine.Model.CommandSpec
@CommandLine.Command(
name = "rbcs", versionProvider = RemoteBuildCacheServerCli.VersionProvider::class
)
class RemoteBuildCacheServerCli : RbcsCommand() {
class VersionProvider : AbstractVersionProvider()
companion object {
@JvmStatic
fun main(vararg args: String) {
val currentClassLoader = RemoteBuildCacheServerCli::class.java.classLoader
Thread.currentThread().contextClassLoader = currentClassLoader
if(currentClassLoader.javaClass.name == "net.woggioni.envelope.loader.ModuleClassLoader") {
//We're running in an envelope jar and custom URL protocols won't work
RbcsUrlStreamHandlerFactory.install()
}
val log = contextLogger()
val app = Application.builder("rbcs")
.configurationDirectoryEnvVar("RBCS_CONFIGURATION_DIR")
.configurationDirectoryPropertyKey("net.woggioni.rbcs.conf.dir")
.build()
val rbcsCli = RemoteBuildCacheServerCli()
val commandLine = CommandLine(rbcsCli)
commandLine.setExecutionExceptionHandler { ex, cl, parseResult ->
log.error(ex.message, ex)
CommandLine.ExitCode.SOFTWARE
}
commandLine.addSubcommand(ServerCommand(app))
commandLine.addSubcommand(PasswordHashCommand())
commandLine.addSubcommand(
CommandLine(ClientCommand(app)).apply {
addSubcommand(BenchmarkCommand())
addSubcommand(PutCommand())
addSubcommand(GetCommand())
addSubcommand(HealthCheckCommand())
})
System.exit(commandLine.execute(*args))
}
}
@CommandLine.Option(names = ["-V", "--version"], versionHelp = true)
var versionHelp = false
private set
@CommandLine.Spec
private lateinit var spec: CommandSpec
override fun run() {
spec.commandLine().usage(System.out);
}
}

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.cli.impl
package net.woggioni.rbcs.cli.impl
import picocli.CommandLine
import java.util.jar.Attributes

View File

@@ -1,11 +1,11 @@
package net.woggioni.gbcs.cli.impl
package net.woggioni.rbcs.cli.impl
import net.woggioni.jwo.Application
import picocli.CommandLine
import java.nio.file.Path
abstract class GbcsCommand : Runnable {
abstract class RbcsCommand : Runnable {
@CommandLine.Option(names = ["-h", "--help"], usageHelp = true)
var usageHelp = false

View File

@@ -0,0 +1,142 @@
package net.woggioni.rbcs.cli.impl.commands
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.client.RemoteBuildCacheClient
import net.woggioni.rbcs.common.contextLogger
import net.woggioni.rbcs.common.error
import net.woggioni.rbcs.common.info
import net.woggioni.jwo.JWO
import picocli.CommandLine
import java.security.SecureRandom
import java.time.Duration
import java.time.Instant
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.Semaphore
import java.util.concurrent.atomic.AtomicLong
import kotlin.random.Random
@CommandLine.Command(
name = "benchmark",
description = ["Run a load test against the server"],
showDefaultValues = true
)
class BenchmarkCommand : RbcsCommand() {
private val log = contextLogger()
@CommandLine.Spec
private lateinit var spec: CommandLine.Model.CommandSpec
@CommandLine.Option(
names = ["-e", "--entries"],
description = ["Total number of elements to be added to the cache"],
paramLabel = "NUMBER_OF_ENTRIES"
)
private var numberOfEntries = 1000
@CommandLine.Option(
names = ["-s", "--size"],
description = ["Size of a cache value in bytes"],
paramLabel = "SIZE"
)
private var size = 0x1000
override fun run() {
val clientCommand = spec.parent().userObject() as ClientCommand
val profile = clientCommand.profileName.let { profileName ->
clientCommand.configuration.profiles[profileName]
?: throw IllegalArgumentException("Profile $profileName does not exist in configuration")
}
RemoteBuildCacheClient(profile).use { client ->
val entryGenerator = sequence {
val random = Random(SecureRandom.getInstance("NativePRNGNonBlocking").nextLong())
while (true) {
val key = JWO.bytesToHex(random.nextBytes(16))
val content = random.nextInt().toByte()
val value = ByteArray(size, { _ -> content })
yield(key to value)
}
}
log.info {
"Starting insertion"
}
val entries = let {
val completionCounter = AtomicLong(0)
val completionQueue = LinkedBlockingQueue<Pair<String, ByteArray>>(numberOfEntries)
val start = Instant.now()
val semaphore = Semaphore(profile.maxConnections * 3)
val iterator = entryGenerator.take(numberOfEntries).iterator()
while (completionCounter.get() < numberOfEntries) {
if (iterator.hasNext()) {
val entry = iterator.next()
semaphore.acquire()
val future = client.put(entry.first, entry.second).thenApply { entry }
future.whenComplete { result, ex ->
if (ex != null) {
log.error(ex.message, ex)
} else {
completionQueue.put(result)
}
semaphore.release()
completionCounter.incrementAndGet()
}
} else {
Thread.sleep(0)
}
}
val inserted = completionQueue.toList()
val end = Instant.now()
log.info {
val elapsed = Duration.between(start, end).toMillis()
val opsPerSecond = String.format("%.2f", numberOfEntries.toDouble() / elapsed * 1000)
"Insertion rate: $opsPerSecond ops/s"
}
inserted
}
log.info {
"Inserted ${entries.size} entries"
}
log.info {
"Starting retrieval"
}
if (entries.isNotEmpty()) {
val completionCounter = AtomicLong(0)
val semaphore = Semaphore(profile.maxConnections * 3)
val start = Instant.now()
val it = entries.iterator()
while (completionCounter.get() < entries.size) {
if (it.hasNext()) {
val entry = it.next()
val future = client.get(entry.first).thenApply {
if (it == null) {
log.error {
"Missing entry for key '${entry.first}'"
}
} else if (!entry.second.contentEquals(it)) {
log.error {
"Retrieved a value different from what was inserted for key '${entry.first}'"
}
}
}
future.whenComplete { _, _ ->
completionCounter.incrementAndGet()
semaphore.release()
}
} else {
Thread.sleep(0)
}
}
val end = Instant.now()
log.info {
val elapsed = Duration.between(start, end).toMillis()
val opsPerSecond = String.format("%.2f", entries.size.toDouble() / elapsed * 1000)
"Retrieval rate: $opsPerSecond ops/s"
}
} else {
log.error("Skipping retrieval benchmark as it was not possible to insert any entry in the cache")
}
}
}
}

View File

@@ -1,24 +1,24 @@
package net.woggioni.gbcs.cli.impl.commands
package net.woggioni.rbcs.cli.impl.commands
import net.woggioni.gbcs.cli.impl.GbcsCommand
import net.woggioni.gbcs.client.GradleBuildCacheClient
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.client.RemoteBuildCacheClient
import net.woggioni.jwo.Application
import picocli.CommandLine
import java.nio.file.Path
@CommandLine.Command(
name = "client",
description = ["GBCS client"],
description = ["RBCS client"],
showDefaultValues = true
)
class ClientCommand(app : Application) : GbcsCommand() {
class ClientCommand(app : Application) : RbcsCommand() {
@CommandLine.Option(
names = ["-c", "--configuration"],
description = ["Path to the client configuration file"],
paramLabel = "CONFIGURATION_FILE"
)
private var configurationFile : Path = findConfigurationFile(app, "gbcs-client.xml")
private var configurationFile : Path = findConfigurationFile(app, "rbcs-client.xml")
@CommandLine.Option(
names = ["-p", "--profile"],
@@ -28,8 +28,8 @@ class ClientCommand(app : Application) : GbcsCommand() {
)
var profileName : String? = null
val configuration : GradleBuildCacheClient.Configuration by lazy {
GradleBuildCacheClient.Configuration.parse(configurationFile)
val configuration : RemoteBuildCacheClient.Configuration by lazy {
RemoteBuildCacheClient.Configuration.parse(configurationFile)
}
override fun run() {

View File

@@ -1,8 +1,8 @@
package net.woggioni.gbcs.cli.impl.commands
package net.woggioni.rbcs.cli.impl.commands
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.cli.impl.GbcsCommand
import net.woggioni.gbcs.client.GradleBuildCacheClient
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.client.RemoteBuildCacheClient
import net.woggioni.rbcs.common.contextLogger
import picocli.CommandLine
import java.nio.file.Files
import java.nio.file.Path
@@ -12,7 +12,7 @@ import java.nio.file.Path
description = ["Fetch a value from the cache with the specified key"],
showDefaultValues = true
)
class GetCommand : GbcsCommand() {
class GetCommand : RbcsCommand() {
private val log = contextLogger()
@CommandLine.Spec
@@ -38,7 +38,7 @@ class GetCommand : GbcsCommand() {
clientCommand.configuration.profiles[profileName]
?: throw IllegalArgumentException("Profile $profileName does not exist in configuration")
}
GradleBuildCacheClient(profile).use { client ->
RemoteBuildCacheClient(profile).use { client ->
client.get(key).thenApply { value ->
value?.let {
(output?.let(Files::newOutputStream) ?: System.out).use {

View File

@@ -0,0 +1,45 @@
package net.woggioni.rbcs.cli.impl.commands
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.client.RemoteBuildCacheClient
import net.woggioni.rbcs.common.contextLogger
import picocli.CommandLine
import java.security.SecureRandom
import kotlin.random.Random
@CommandLine.Command(
name = "health",
description = ["Check server health"],
showDefaultValues = true
)
class HealthCheckCommand : RbcsCommand() {
private val log = contextLogger()
@CommandLine.Spec
private lateinit var spec: CommandLine.Model.CommandSpec
override fun run() {
val clientCommand = spec.parent().userObject() as ClientCommand
val profile = clientCommand.profileName.let { profileName ->
clientCommand.configuration.profiles[profileName]
?: throw IllegalArgumentException("Profile $profileName does not exist in configuration")
}
RemoteBuildCacheClient(profile).use { client ->
val random = Random(SecureRandom.getInstance("NativePRNGNonBlocking").nextLong())
val nonce = ByteArray(0xa0)
random.nextBytes(nonce)
client.healthCheck(nonce).thenApply { value ->
if(value == null) {
throw IllegalStateException("Empty response from server")
}
for(i in 0 until nonce.size) {
for(j in value.size - nonce.size until nonce.size) {
if(nonce[i] != value[j]) {
throw IllegalStateException("Server nonce does not match")
}
}
}
}.get()
}
}
}

View File

@@ -1,8 +1,8 @@
package net.woggioni.gbcs.cli.impl.commands
package net.woggioni.rbcs.cli.impl.commands
import net.woggioni.gbcs.common.PasswordSecurity.hashPassword
import net.woggioni.gbcs.cli.impl.GbcsCommand
import net.woggioni.gbcs.cli.impl.converters.OutputStreamConverter
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.cli.impl.converters.OutputStreamConverter
import net.woggioni.rbcs.common.PasswordSecurity.hashPassword
import net.woggioni.jwo.UncloseableOutputStream
import picocli.CommandLine
import java.io.OutputStream
@@ -12,10 +12,10 @@ import java.io.PrintWriter
@CommandLine.Command(
name = "password",
description = ["Generate a password hash to add to GBCS configuration file"],
description = ["Generate a password hash to add to RBCS configuration file"],
showDefaultValues = true
)
class PasswordHashCommand : GbcsCommand() {
class PasswordHashCommand : RbcsCommand() {
@CommandLine.Option(
names = ["-o", "--output-file"],
description = ["Write the output to a file instead of stdout"],

View File

@@ -1,9 +1,9 @@
package net.woggioni.gbcs.cli.impl.commands
package net.woggioni.rbcs.cli.impl.commands
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.cli.impl.GbcsCommand
import net.woggioni.gbcs.cli.impl.converters.InputStreamConverter
import net.woggioni.gbcs.client.GradleBuildCacheClient
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.cli.impl.converters.InputStreamConverter
import net.woggioni.rbcs.client.RemoteBuildCacheClient
import net.woggioni.rbcs.common.contextLogger
import picocli.CommandLine
import java.io.InputStream
@@ -12,7 +12,7 @@ import java.io.InputStream
description = ["Add or replace a value to the cache with the specified key"],
showDefaultValues = true
)
class PutCommand : GbcsCommand() {
class PutCommand : RbcsCommand() {
private val log = contextLogger()
@CommandLine.Spec
@@ -39,7 +39,7 @@ class PutCommand : GbcsCommand() {
clientCommand.configuration.profiles[profileName]
?: throw IllegalArgumentException("Profile $profileName does not exist in configuration")
}
GradleBuildCacheClient(profile).use { client ->
RemoteBuildCacheClient(profile).use { client ->
value.use {
client.put(key, it.readAllBytes())
}.get()

View File

@@ -1,25 +1,26 @@
package net.woggioni.gbcs.cli.impl.commands
package net.woggioni.rbcs.cli.impl.commands
import net.woggioni.gbcs.server.GradleBuildCacheServer
import net.woggioni.gbcs.server.GradleBuildCacheServer.Companion.DEFAULT_CONFIGURATION_URL
import net.woggioni.gbcs.api.Configuration
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.common.debug
import net.woggioni.gbcs.common.info
import net.woggioni.gbcs.cli.impl.GbcsCommand
import net.woggioni.rbcs.cli.impl.RbcsCommand
import net.woggioni.rbcs.cli.impl.converters.DurationConverter
import net.woggioni.rbcs.common.contextLogger
import net.woggioni.rbcs.common.debug
import net.woggioni.rbcs.common.info
import net.woggioni.rbcs.server.RemoteBuildCacheServer
import net.woggioni.rbcs.server.RemoteBuildCacheServer.Companion.DEFAULT_CONFIGURATION_URL
import net.woggioni.jwo.Application
import net.woggioni.jwo.JWO
import picocli.CommandLine
import java.io.ByteArrayOutputStream
import java.nio.file.Files
import java.nio.file.Path
import java.time.Duration
@CommandLine.Command(
name = "server",
description = ["GBCS server"],
description = ["RBCS server"],
showDefaultValues = true
)
class ServerCommand(app : Application) : GbcsCommand() {
class ServerCommand(app : Application) : RbcsCommand() {
private val log = contextLogger()
@@ -35,16 +36,20 @@ class ServerCommand(app : Application) : GbcsCommand() {
}
}
@CommandLine.Option(
names = ["-t", "--timeout"],
description = ["Exit after the specified time"],
paramLabel = "TIMEOUT",
converter = [DurationConverter::class]
)
private var timeout: Duration? = null
@CommandLine.Option(
names = ["-c", "--config-file"],
description = ["Read the application configuration from this file"],
paramLabel = "CONFIG_FILE"
)
private var configurationFile: Path = findConfigurationFile(app, "gbcs-server.xml")
val configuration : Configuration by lazy {
GradleBuildCacheServer.loadConfiguration(configurationFile)
}
private var configurationFile: Path = findConfigurationFile(app, "rbcs-server.xml")
override fun run() {
if (!Files.exists(configurationFile)) {
@@ -52,16 +57,20 @@ class ServerCommand(app : Application) : GbcsCommand() {
createDefaultConfigurationFile(configurationFile)
}
val configuration = GradleBuildCacheServer.loadConfiguration(configurationFile)
val configuration = RemoteBuildCacheServer.loadConfiguration(configurationFile)
log.debug {
ByteArrayOutputStream().also {
GradleBuildCacheServer.dumpConfiguration(configuration, it)
RemoteBuildCacheServer.dumpConfiguration(configuration, it)
}.let {
"Server configuration:\n${String(it.toByteArray())}"
}
}
val server = GradleBuildCacheServer(configuration)
server.run().use {
val server = RemoteBuildCacheServer(configuration)
server.run().use { server ->
timeout?.let {
Thread.sleep(it)
server.shutdown()
}
}
}
}

View File

@@ -0,0 +1,11 @@
package net.woggioni.rbcs.cli.impl.converters
import picocli.CommandLine
import java.time.Duration
class DurationConverter : CommandLine.ITypeConverter<Duration> {
override fun convert(value: String): Duration {
return Duration.parse(value)
}
}

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.cli.impl.converters
package net.woggioni.rbcs.cli.impl.converters
import picocli.CommandLine
import java.io.InputStream

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.cli.impl.converters
package net.woggioni.rbcs.cli.impl.converters
import picocli.CommandLine
import java.io.OutputStream

View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration>
<configuration>
<import class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"/>
<import class="ch.qos.logback.core.ConsoleAppender"/>
<appender name="console" class="ConsoleAppender">
<target>System.err</target>
<encoder class="PatternLayoutEncoder">
<pattern>%d [%highlight(%-5level)] \(%thread\) %logger{36} -%kvp- %msg %n</pattern>
</encoder>
</appender>
<root level="info">
<appender-ref ref="console"/>
</root>
</configuration>

View File

@@ -4,12 +4,14 @@ plugins {
}
dependencies {
implementation project(':gbcs-api')
implementation project(':gbcs-common')
implementation project(':rbcs-api')
implementation project(':rbcs-common')
implementation catalog.picocli
implementation catalog.slf4j.api
implementation catalog.netty.buffer
implementation catalog.netty.codec.http
testRuntimeOnly catalog.logback.classic
}

View File

@@ -1,4 +1,4 @@
module net.woggioni.gbcs.client {
module net.woggioni.rbcs.client {
requires io.netty.handler;
requires io.netty.codec.http;
requires io.netty.transport;
@@ -6,12 +6,12 @@ module net.woggioni.gbcs.client {
requires io.netty.common;
requires io.netty.buffer;
requires java.xml;
requires net.woggioni.gbcs.common;
requires net.woggioni.gbcs.api;
requires net.woggioni.rbcs.common;
requires net.woggioni.rbcs.api;
requires io.netty.codec;
requires org.slf4j;
exports net.woggioni.gbcs.client;
exports net.woggioni.rbcs.client;
opens net.woggioni.gbcs.client.schema;
opens net.woggioni.rbcs.client.schema;
}

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.client
package net.woggioni.rbcs.client
import io.netty.bootstrap.Bootstrap
import io.netty.buffer.ByteBuf
@@ -30,11 +30,11 @@ import io.netty.handler.ssl.SslContextBuilder
import io.netty.handler.stream.ChunkedWriteHandler
import io.netty.util.concurrent.Future
import io.netty.util.concurrent.GenericFutureListener
import net.woggioni.gbcs.client.impl.Parser
import net.woggioni.gbcs.common.Xml
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.common.debug
import net.woggioni.gbcs.common.trace
import net.woggioni.rbcs.client.impl.Parser
import net.woggioni.rbcs.common.Xml
import net.woggioni.rbcs.common.contextLogger
import net.woggioni.rbcs.common.debug
import net.woggioni.rbcs.common.trace
import java.net.InetSocketAddress
import java.net.URI
import java.nio.file.Files
@@ -48,7 +48,7 @@ import java.util.concurrent.atomic.AtomicInteger
import io.netty.util.concurrent.Future as NettyFuture
class GradleBuildCacheClient(private val profile: Configuration.Profile) : AutoCloseable {
class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoCloseable {
private val group: NioEventLoopGroup
private var sslContext: SslContext
private val log = contextLogger()
@@ -66,11 +66,18 @@ class GradleBuildCacheClient(private val profile: Configuration.Profile) : AutoC
data class BasicAuthenticationCredentials(val username: String, val password: String) : Authentication()
}
class RetryPolicy(
val maxAttempts: Int,
val initialDelayMillis: Long,
val exp: Double
)
data class Profile(
val serverURI: URI,
val authentication: Authentication?,
val connectionTimeout: Duration?,
val maxConnections : Int
val maxConnections: Int,
val retryPolicy: RetryPolicy?,
)
companion object {
@@ -161,9 +168,74 @@ class GradleBuildCacheClient(private val profile: Configuration.Profile) : AutoC
pool = FixedChannelPool(bootstrap, channelPoolHandler, profile.maxConnections)
}
private fun executeWithRetry(operation: () -> CompletableFuture<FullHttpResponse>): CompletableFuture<FullHttpResponse> {
val retryPolicy = profile.retryPolicy
return if (retryPolicy != null) {
val outcomeHandler = OutcomeHandler<FullHttpResponse> { outcome ->
when (outcome) {
is OperationOutcome.Success -> {
val response = outcome.result
val status = response.status()
when (status) {
HttpResponseStatus.TOO_MANY_REQUESTS -> {
val retryAfter = response.headers()[HttpHeaderNames.RETRY_AFTER]?.let { headerValue ->
try {
headerValue.toLong() * 1000
} catch (nfe: NumberFormatException) {
null
}
}
OutcomeHandlerResult.Retry(retryAfter)
}
HttpResponseStatus.INTERNAL_SERVER_ERROR, HttpResponseStatus.SERVICE_UNAVAILABLE ->
OutcomeHandlerResult.Retry()
else -> OutcomeHandlerResult.DoNotRetry()
}
}
is OperationOutcome.Failure -> {
OutcomeHandlerResult.Retry()
}
}
}
executeWithRetry(
group,
retryPolicy.maxAttempts,
retryPolicy.initialDelayMillis.toDouble(),
retryPolicy.exp,
outcomeHandler,
operation
)
} else {
operation()
}
}
fun healthCheck(nonce: ByteArray): CompletableFuture<ByteArray?> {
return executeWithRetry {
sendRequest(profile.serverURI, HttpMethod.TRACE, nonce)
}.thenApply {
val status = it.status()
if (it.status() != HttpResponseStatus.OK) {
throw HttpException(status)
} else {
it.content()
}
}.thenApply { maybeByteBuf ->
maybeByteBuf?.let {
val result = ByteArray(it.readableBytes())
it.getBytes(0, result)
result
}
}
}
fun get(key: String): CompletableFuture<ByteArray?> {
return sendRequest(profile.serverURI.resolve(key), HttpMethod.GET, null)
.thenApply {
return executeWithRetry {
sendRequest(profile.serverURI.resolve(key), HttpMethod.GET, null)
}.thenApply {
val status = it.status()
if (it.status() == HttpResponseStatus.NOT_FOUND) {
null
@@ -182,7 +254,9 @@ class GradleBuildCacheClient(private val profile: Configuration.Profile) : AutoC
}
fun put(key: String, content: ByteArray): CompletableFuture<Unit> {
return sendRequest(profile.serverURI.resolve(key), HttpMethod.PUT, content).thenApply {
return executeWithRetry {
sendRequest(profile.serverURI.resolve(key), HttpMethod.PUT, content)
}.thenApply {
val status = it.status()
if (it.status() != HttpResponseStatus.CREATED && it.status() != HttpResponseStatus.OK) {
throw HttpException(status)

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.client
package net.woggioni.rbcs.client
import io.netty.handler.codec.http.HttpResponseStatus

View File

@@ -0,0 +1,108 @@
package net.woggioni.rbcs.client.impl
import net.woggioni.rbcs.api.exception.ConfigurationException
import net.woggioni.rbcs.client.RemoteBuildCacheClient
import net.woggioni.rbcs.common.Xml.Companion.asIterable
import net.woggioni.rbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import java.net.URI
import java.nio.file.Files
import java.nio.file.Path
import java.security.KeyStore
import java.security.PrivateKey
import java.security.cert.X509Certificate
import java.time.Duration
object Parser {
fun parse(document: Document): RemoteBuildCacheClient.Configuration {
val root = document.documentElement
val profiles = mutableMapOf<String, RemoteBuildCacheClient.Configuration.Profile>()
for (child in root.asIterable()) {
val tagName = child.localName
when (tagName) {
"profile" -> {
val name =
child.renderAttribute("name") ?: throw ConfigurationException("name attribute is required")
val uri = child.renderAttribute("base-url")?.let(::URI)
?: throw ConfigurationException("base-url attribute is required")
var authentication: RemoteBuildCacheClient.Configuration.Authentication? = null
var retryPolicy: RemoteBuildCacheClient.Configuration.RetryPolicy? = null
for (gchild in child.asIterable()) {
when (gchild.localName) {
"tls-client-auth" -> {
val keyStoreFile = gchild.renderAttribute("key-store-file")
val keyStorePassword =
gchild.renderAttribute("key-store-password")
val keyAlias = gchild.renderAttribute("key-alias")
val keyPassword = gchild.renderAttribute("key-password")
val keystore = KeyStore.getInstance("PKCS12").apply {
Files.newInputStream(Path.of(keyStoreFile)).use {
load(it, keyStorePassword?.toCharArray())
}
}
val key = keystore.getKey(keyAlias, keyPassword?.toCharArray()) as PrivateKey
val certChain = keystore.getCertificateChain(keyAlias).asSequence()
.map { it as X509Certificate }
.toList()
.toTypedArray()
authentication =
RemoteBuildCacheClient.Configuration.Authentication.TlsClientAuthenticationCredentials(
key,
certChain
)
}
"basic-auth" -> {
val username = gchild.renderAttribute("user")
?: throw ConfigurationException("username attribute is required")
val password = gchild.renderAttribute("password")
?: throw ConfigurationException("password attribute is required")
authentication =
RemoteBuildCacheClient.Configuration.Authentication.BasicAuthenticationCredentials(
username,
password
)
}
"retry-policy" -> {
val maxAttempts =
gchild.renderAttribute("max-attempts")
?.let(String::toInt)
?: throw ConfigurationException("max-attempts attribute is required")
val initialDelay =
gchild.renderAttribute("initial-delay")
?.let(Duration::parse)
?: Duration.ofSeconds(1)
val exp =
gchild.renderAttribute("exp")
?.let(String::toDouble)
?: 2.0f
retryPolicy = RemoteBuildCacheClient.Configuration.RetryPolicy(
maxAttempts,
initialDelay.toMillis(),
exp.toDouble()
)
}
}
}
val maxConnections = child.renderAttribute("max-connections")
?.let(String::toInt)
?: 50
val connectionTimeout = child.renderAttribute("connection-timeout")
?.let(Duration::parse)
profiles[name] = RemoteBuildCacheClient.Configuration.Profile(
uri,
authentication,
connectionTimeout,
maxConnections,
retryPolicy
)
}
}
}
return RemoteBuildCacheClient.Configuration(profiles)
}
}

View File

@@ -0,0 +1,75 @@
package net.woggioni.rbcs.client
import io.netty.util.concurrent.EventExecutorGroup
import java.util.concurrent.CompletableFuture
import java.util.concurrent.TimeUnit
sealed class OperationOutcome<T> {
class Success<T>(val result: T) : OperationOutcome<T>()
class Failure<T>(val ex: Throwable) : OperationOutcome<T>()
}
sealed class OutcomeHandlerResult {
class Retry(val suggestedDelayMillis: Long? = null) : OutcomeHandlerResult()
class DoNotRetry : OutcomeHandlerResult()
}
fun interface OutcomeHandler<T> {
fun shouldRetry(result: OperationOutcome<T>): OutcomeHandlerResult
}
fun <T> executeWithRetry(
eventExecutorGroup: EventExecutorGroup,
maxAttempts: Int,
initialDelay: Double,
exp: Double,
outcomeHandler: OutcomeHandler<T>,
cb: () -> CompletableFuture<T>
): CompletableFuture<T> {
val finalResult = cb()
var future = finalResult
var shortCircuit = false
for (i in 1 until maxAttempts) {
future = future.handle { result, ex ->
val operationOutcome = if (ex == null) {
OperationOutcome.Success(result)
} else {
OperationOutcome.Failure(ex.cause ?: ex)
}
if (shortCircuit) {
when(operationOutcome) {
is OperationOutcome.Failure -> throw operationOutcome.ex
is OperationOutcome.Success -> CompletableFuture.completedFuture(operationOutcome.result)
}
} else {
when(val outcomeHandlerResult = outcomeHandler.shouldRetry(operationOutcome)) {
is OutcomeHandlerResult.Retry -> {
val res = CompletableFuture<T>()
val delay = run {
val scheduledDelay = (initialDelay * Math.pow(exp, i.toDouble())).toLong()
outcomeHandlerResult.suggestedDelayMillis?.coerceAtMost(scheduledDelay) ?: scheduledDelay
}
eventExecutorGroup.schedule({
cb().handle { result, ex ->
if (ex == null) {
res.complete(result)
} else {
res.completeExceptionally(ex)
}
}
}, delay, TimeUnit.MILLISECONDS)
res
}
is OutcomeHandlerResult.DoNotRetry -> {
shortCircuit = true
when(operationOutcome) {
is OperationOutcome.Failure -> throw operationOutcome.ex
is OperationOutcome.Success -> CompletableFuture.completedFuture(operationOutcome.result)
}
}
}
}
}.thenCompose { it }
}
return future
}

View File

@@ -1,23 +1,26 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<xs:schema targetNamespace="urn:net.woggioni.gbcs.client"
<xs:schema targetNamespace="urn:net.woggioni.rbcs.client"
xmlns:xs="http://www.w3.org/2001/XMLSchema"
xmlns:gbcs-client="urn:net.woggioni.gbcs.client"
xmlns:rbcs-client="urn:net.woggioni.rbcs.client"
elementFormDefault="unqualified"
>
<xs:element name="profiles" type="gbcs-client:profilesType"/>
<xs:element name="profiles" type="rbcs-client:profilesType"/>
<xs:complexType name="profilesType">
<xs:sequence minOccurs="0">
<xs:element name="profile" type="gbcs-client:profileType" maxOccurs="unbounded"/>
<xs:element name="profile" type="rbcs-client:profileType" maxOccurs="unbounded"/>
</xs:sequence>
</xs:complexType>
<xs:complexType name="profileType">
<xs:sequence>
<xs:choice>
<xs:element name="no-auth" type="gbcs-client:noAuthType"/>
<xs:element name="basic-auth" type="gbcs-client:basicAuthType"/>
<xs:element name="tls-client-auth" type="gbcs-client:tlsClientAuthType"/>
<xs:element name="no-auth" type="rbcs-client:noAuthType"/>
<xs:element name="basic-auth" type="rbcs-client:basicAuthType"/>
<xs:element name="tls-client-auth" type="rbcs-client:tlsClientAuthType"/>
</xs:choice>
<xs:element name="retry-policy" type="rbcs-client:retryType" minOccurs="0"/>
</xs:sequence>
<xs:attribute name="name" type="xs:token" use="required"/>
<xs:attribute name="base-url" type="xs:anyURI" use="required"/>
<xs:attribute name="max-connections" type="xs:positiveInteger" default="50"/>
@@ -38,4 +41,10 @@
<xs:attribute name="key-password" type="xs:string" use="optional"/>
</xs:complexType>
<xs:complexType name="retryType">
<xs:attribute name="max-attempts" type="xs:positiveInteger" use="required"/>
<xs:attribute name="initial-delay" type="xs:duration" default="PT1S"/>
<xs:attribute name="exp" type="xs:double" default="2.0"/>
</xs:complexType>
</xs:schema>

View File

@@ -0,0 +1,148 @@
package net.woggioni.rbcs.client
import io.netty.util.concurrent.DefaultEventExecutorGroup
import io.netty.util.concurrent.EventExecutorGroup
import net.woggioni.rbcs.common.contextLogger
import org.junit.jupiter.api.Assertions
import org.junit.jupiter.api.extension.ExtensionContext
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments
import org.junit.jupiter.params.provider.ArgumentsProvider
import org.junit.jupiter.params.provider.ArgumentsSource
import java.util.concurrent.CompletableFuture
import java.util.stream.Stream
import kotlin.random.Random
class RetryTest {
data class TestArgs(
val seed: Int,
val maxAttempt: Int,
val initialDelay: Double,
val exp: Double,
)
class TestArguments : ArgumentsProvider {
override fun provideArguments(context: ExtensionContext): Stream<out Arguments> {
return Stream.of(
TestArgs(
seed = 101325,
maxAttempt = 5,
initialDelay = 50.0,
exp = 2.0,
),
TestArgs(
seed = 101325,
maxAttempt = 20,
initialDelay = 100.0,
exp = 1.1,
),
TestArgs(
seed = 123487,
maxAttempt = 20,
initialDelay = 100.0,
exp = 2.0,
),
TestArgs(
seed = 20082024,
maxAttempt = 10,
initialDelay = 100.0,
exp = 2.0,
)
).map {
object: Arguments {
override fun get() = arrayOf(it)
}
}
}
}
@ArgumentsSource(TestArguments::class)
@ParameterizedTest
fun test(testArgs: TestArgs) {
val log = contextLogger()
log.debug("Start")
val executor: EventExecutorGroup = DefaultEventExecutorGroup(1)
val attempts = mutableListOf<Pair<Long, OperationOutcome<Int>>>()
val outcomeHandler = OutcomeHandler<Int> { outcome ->
when(outcome) {
is OperationOutcome.Success -> {
if(outcome.result % 10 == 0) {
OutcomeHandlerResult.DoNotRetry()
} else {
OutcomeHandlerResult.Retry(null)
}
}
is OperationOutcome.Failure -> {
when(outcome.ex) {
is IllegalStateException -> {
log.debug(outcome.ex.message, outcome.ex)
OutcomeHandlerResult.Retry(null)
}
else -> {
OutcomeHandlerResult.DoNotRetry()
}
}
}
}
}
val random = Random(testArgs.seed)
val future =
executeWithRetry(executor, testArgs.maxAttempt, testArgs.initialDelay, testArgs.exp, outcomeHandler) {
val now = System.nanoTime()
val result = CompletableFuture<Int>()
executor.submit {
val n = random.nextInt(0, Integer.MAX_VALUE)
log.debug("Got new number: {}", n)
if(n % 3 == 0) {
val ex = IllegalStateException("Value $n can be divided by 3")
result.completeExceptionally(ex)
attempts += now to OperationOutcome.Failure(ex)
} else if(n % 7 == 0) {
val ex = RuntimeException("Value $n can be divided by 7")
result.completeExceptionally(ex)
attempts += now to OperationOutcome.Failure(ex)
} else {
result.complete(n)
attempts += now to OperationOutcome.Success(n)
}
}
result
}
Assertions.assertTrue(attempts.size <= testArgs.maxAttempt)
val result = future.handle { res, ex ->
if(ex != null) {
val err = ex.cause ?: ex
log.debug(err.message, err)
OperationOutcome.Failure(err)
} else {
OperationOutcome.Success(res)
}
}.get()
for ((index, attempt) in attempts.withIndex()) {
val (timestamp, value) = attempt
if (index > 0) {
/* Check the delay for subsequent attempts is correct */
val previousAttempt = attempts[index - 1]
val expectedTimestamp =
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)
}
if (index == attempts.size - 1 && index < testArgs.maxAttempt - 1) {
/*
* If the last attempt index is lower than the maximum number of attempts, then
* check the outcome handler returns DoNotRetry
*/
Assertions.assertTrue(outcomeHandler.shouldRetry(value) is OutcomeHandlerResult.DoNotRetry)
} else if (index < attempts.size - 1) {
/*
* If the attempt is not the last attempt check the outcome handler returns Retry
*/
Assertions.assertTrue(outcomeHandler.shouldRetry(value) is OutcomeHandlerResult.Retry)
}
}
}
}

View File

@@ -15,6 +15,7 @@
<root level="info">
<appender-ref ref="console"/>
</root>
<logger name="io.netty" level="info"/>
<logger name="com.google.code.yanf4j" level="warn"/>
<logger name="net.rubyeye.xmemcached" level="warn"/>
</configuration>

View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<rbcs-client:profiles xmlns:xs="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rbcs-client="urn:net.woggioni.rbcs.client"
xs:schemaLocation="urn:net.woggioni.rbcs.client jms://net.woggioni.rbcs.client/net/woggioni/rbcs/client/schema/rbcs-client.xsd"
>
<profile name="profile1" base-url="https://rbcs1.example.com/">
<tls-client-auth
key-store-file="keystore.pfx"
key-store-password="password"
key-alias="woggioni@c962475fa38"
key-password="key-password"/>
</profile>
<profile name="profile2" base-url="https://rbcs2.example.com/">
<basic-auth user="user" password="password"/>
</profile>
</rbcs-client:profiles>

View File

@@ -6,9 +6,10 @@ plugins {
}
dependencies {
implementation project(':gbcs-api')
implementation project(':rbcs-api')
implementation catalog.slf4j.api
implementation catalog.jwo
implementation catalog.netty.buffer
}
publishing {

View File

@@ -1,10 +1,11 @@
module net.woggioni.gbcs.common {
module net.woggioni.rbcs.common {
requires java.xml;
requires java.logging;
requires org.slf4j;
requires kotlin.stdlib;
requires net.woggioni.jwo;
requires io.netty.buffer;
provides java.net.spi.URLStreamHandlerProvider with net.woggioni.gbcs.common.GbcsUrlStreamHandlerFactory;
exports net.woggioni.gbcs.common;
provides java.net.spi.URLStreamHandlerProvider with net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory;
exports net.woggioni.rbcs.common;
}

View File

@@ -0,0 +1,25 @@
package net.woggioni.rbcs.common
import io.netty.buffer.ByteBuf
import java.io.InputStream
class ByteBufInputStream(private val buf : ByteBuf) : InputStream() {
override fun read(): Int {
return buf.takeIf {
it.readableBytes() > 0
}?.let(ByteBuf::readByte)
?.let(Byte::toInt) ?: -1
}
override fun read(b: ByteArray, off: Int, len: Int): Int {
val readableBytes = buf.readableBytes()
if(readableBytes == 0) return -1
val result = len.coerceAtMost(readableBytes)
buf.readBytes(b, off, result)
return result
}
override fun close() {
buf.release()
}
}

View File

@@ -0,0 +1,18 @@
package net.woggioni.rbcs.common
import io.netty.buffer.ByteBuf
import java.io.OutputStream
class ByteBufOutputStream(private val buf : ByteBuf) : OutputStream() {
override fun write(b: Int) {
buf.writeByte(b)
}
override fun write(b: ByteArray, off: Int, len: Int) {
buf.writeBytes(b, off, len)
}
override fun close() {
buf.release()
}
}

View File

@@ -0,0 +1,7 @@
package net.woggioni.rbcs.common
class ResourceNotFoundException(msg : String? = null, cause: Throwable? = null) : RuntimeException(msg, cause) {
}
class ModuleNotFoundException(msg : String? = null, cause: Throwable? = null) : RuntimeException(msg, cause) {
}

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.common
package net.woggioni.rbcs.common
data class HostAndPort(val host: String, val port: Int = 0) {

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.common
package net.woggioni.rbcs.common
import org.slf4j.Logger
import org.slf4j.LoggerFactory

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.common
package net.woggioni.rbcs.common
import java.security.SecureRandom
import java.security.spec.KeySpec

View File

@@ -1,9 +1,17 @@
package net.woggioni.gbcs.server.cache
package net.woggioni.rbcs.common
import net.woggioni.jwo.JWO
import java.net.URI
import java.net.URL
import java.security.MessageDigest
object CacheUtils {
object RBCS {
fun String.toUrl() : URL = URL.of(URI(this), null)
const val RBCS_NAMESPACE_URI: String = "urn:net.woggioni.rbcs.server"
const val RBCS_PREFIX: String = "rbcs"
const val XML_SCHEMA_NAMESPACE_URI = "http://www.w3.org/2001/XMLSchema-instance"
fun digest(
data: ByteArray,
md: MessageDigest = MessageDigest.getInstance("MD5")

View File

@@ -1,20 +1,18 @@
package net.woggioni.gbcs.common
package net.woggioni.rbcs.common
import java.io.IOException
import java.io.InputStream
import java.net.URL
import java.net.URLConnection
import java.net.URLStreamHandler
import java.net.URLStreamHandlerFactory
import java.net.spi.URLStreamHandlerProvider
import java.util.Optional
import java.util.concurrent.atomic.AtomicBoolean
import java.util.stream.Collectors
class GbcsUrlStreamHandlerFactory : URLStreamHandlerProvider() {
class RbcsUrlStreamHandlerFactory : URLStreamHandlerProvider() {
private class ClasspathHandler(private val classLoader: ClassLoader = GbcsUrlStreamHandlerFactory::class.java.classLoader) :
private class ClasspathHandler(private val classLoader: ClassLoader = RbcsUrlStreamHandlerFactory::class.java.classLoader) :
URLStreamHandler() {
override fun openConnection(u: URL): URLConnection? {
@@ -37,13 +35,17 @@ class GbcsUrlStreamHandlerFactory : URLStreamHandlerProvider() {
private class JpmsHandler : URLStreamHandler() {
override fun openConnection(u: URL): URLConnection {
val thisModule = javaClass.module
val sourceModule = Optional.ofNullable(thisModule)
.map { obj: Module -> obj.layer }
.flatMap { layer: ModuleLayer ->
val moduleName = u.host
layer.findModule(moduleName)
}.orElse(thisModule)
val thisModule = javaClass.module
val sourceModule =
thisModule
?.let(Module::getLayer)
?.let { layer: ModuleLayer ->
layer.findModule(moduleName).orElse(null)
} ?: if(thisModule.layer == null) {
thisModule
} else throw ModuleNotFoundException("Module '$moduleName' not found")
return JpmsResourceURLConnection(u, sourceModule)
}
}
@@ -54,7 +56,9 @@ class GbcsUrlStreamHandlerFactory : URLStreamHandlerProvider() {
@Throws(IOException::class)
override fun getInputStream(): InputStream {
return module.getResourceAsStream(getURL().path)
val resource = getURL().path
return module.getResourceAsStream(resource)
?: throw ResourceNotFoundException("Resource '$resource' not found in module '${module.name}'")
}
}
@@ -83,12 +87,12 @@ class GbcsUrlStreamHandlerFactory : URLStreamHandlerProvider() {
private val installed = AtomicBoolean(false)
fun install() {
if (!installed.getAndSet(true)) {
URL.setURLStreamHandlerFactory(GbcsUrlStreamHandlerFactory())
URL.setURLStreamHandlerFactory(RbcsUrlStreamHandlerFactory())
}
}
private val packageMap: Map<String, List<Module>> by lazy {
GbcsUrlStreamHandlerFactory::class.java.module.layer
RbcsUrlStreamHandlerFactory::class.java.module.layer
.modules()
.stream()
.flatMap { m: Module ->

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.common
package net.woggioni.rbcs.common
import net.woggioni.jwo.JWO
import org.slf4j.LoggerFactory

View File

@@ -0,0 +1 @@
net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory

View File

@@ -6,10 +6,10 @@ plugins {
configurations {
bundle {
extendsFrom runtimeClasspath
canBeResolved = true
canBeConsumed = false
visible = false
transitive = false
resolutionStrategy {
dependencies {
@@ -29,10 +29,20 @@ configurations {
}
dependencies {
compileOnly project(':gbcs-common')
compileOnly project(':gbcs-api')
compileOnly catalog.jwo
implementation catalog.xmemcached
implementation project(':rbcs-common')
implementation project(':rbcs-api')
implementation catalog.jwo
implementation catalog.slf4j.api
implementation catalog.netty.common
implementation catalog.netty.codec.memcache
bundle catalog.netty.codec.memcache
testRuntimeOnly catalog.logback.classic
}
tasks.named(JavaPlugin.TEST_TASK_NAME, Test) {
systemProperty("io.netty.leakDetectionLevel", "PARANOID")
}
Provider<Tar> bundleTask = tasks.register("bundle", Tar) {

View File

@@ -0,0 +1,19 @@
import net.woggioni.rbcs.api.CacheProvider;
module net.woggioni.rbcs.server.memcache {
requires net.woggioni.rbcs.common;
requires net.woggioni.rbcs.api;
requires net.woggioni.jwo;
requires java.xml;
requires kotlin.stdlib;
requires io.netty.transport;
requires io.netty.codec;
requires io.netty.codec.memcache;
requires io.netty.common;
requires io.netty.buffer;
requires org.slf4j;
provides CacheProvider with net.woggioni.rbcs.server.memcache.MemcacheCacheProvider;
opens net.woggioni.rbcs.server.memcache.schema;
}

View File

@@ -0,0 +1,4 @@
package net.woggioni.rbcs.server.memcache
class MemcacheException(status : Short, msg : String? = null, cause : Throwable? = null)
: RuntimeException(msg ?: "Memcached status $status", cause)

View File

@@ -0,0 +1,23 @@
package net.woggioni.rbcs.server.memcache
import io.netty.buffer.ByteBuf
import net.woggioni.rbcs.api.Cache
import net.woggioni.rbcs.server.memcache.client.MemcacheClient
import java.nio.channels.ReadableByteChannel
import java.util.concurrent.CompletableFuture
class MemcacheCache(private val cfg : MemcacheCacheConfiguration) : Cache {
private val memcacheClient = MemcacheClient(cfg)
override fun get(key: String): CompletableFuture<ReadableByteChannel?> {
return memcacheClient.get(key)
}
override fun put(key: String, content: ByteBuf): CompletableFuture<Void> {
return memcacheClient.put(key, content, cfg.maxAge)
}
override fun close() {
memcacheClient.close()
}
}

View File

@@ -0,0 +1,40 @@
package net.woggioni.rbcs.server.memcache
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.common.HostAndPort
import java.time.Duration
data class MemcacheCacheConfiguration(
val servers: List<Server>,
val maxAge: Duration = Duration.ofDays(1),
val maxSize: Int = 0x100000,
val digestAlgorithm: String? = null,
val compressionMode: CompressionMode? = null,
) : Configuration.Cache {
enum class CompressionMode {
/**
* Gzip mode
*/
GZIP,
/**
* Deflate mode
*/
DEFLATE
}
data class Server(
val endpoint : HostAndPort,
val connectionTimeoutMillis : Int?,
val maxConnections : Int
)
override fun materialize() = MemcacheCache(this)
override fun getNamespaceURI() = "urn:net.woggioni.rbcs.server.memcache"
override fun getTypeName() = "memcacheCacheType"
}

View File

@@ -0,0 +1,101 @@
package net.woggioni.rbcs.server.memcache
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.api.exception.ConfigurationException
import net.woggioni.rbcs.common.RBCS
import net.woggioni.rbcs.common.HostAndPort
import net.woggioni.rbcs.common.Xml
import net.woggioni.rbcs.common.Xml.Companion.asIterable
import net.woggioni.rbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import org.w3c.dom.Element
import java.time.Duration
import java.time.temporal.ChronoUnit
class MemcacheCacheProvider : CacheProvider<MemcacheCacheConfiguration> {
override fun getXmlSchemaLocation() = "jpms://net.woggioni.rbcs.server.memcache/net/woggioni/rbcs/server/memcache/schema/rbcs-memcache.xsd"
override fun getXmlType() = "memcacheCacheType"
override fun getXmlNamespace() = "urn:net.woggioni.rbcs.server.memcache"
val xmlNamespacePrefix : String
get() = "rbcs-memcache"
override fun deserialize(el: Element): MemcacheCacheConfiguration {
val servers = mutableListOf<MemcacheCacheConfiguration.Server>()
val maxAge = el.renderAttribute("max-age")
?.let(Duration::parse)
?: Duration.ofDays(1)
val maxSize = el.renderAttribute("max-size")
?.let(String::toInt)
?: 0x100000
val compressionMode = el.renderAttribute("compression-mode")
?.let {
when (it) {
"gzip" -> MemcacheCacheConfiguration.CompressionMode.GZIP
"deflate" -> MemcacheCacheConfiguration.CompressionMode.DEFLATE
else -> MemcacheCacheConfiguration.CompressionMode.DEFLATE
}
}
?: MemcacheCacheConfiguration.CompressionMode.DEFLATE
val digestAlgorithm = el.renderAttribute("digest")
for (child in el.asIterable()) {
when (child.nodeName) {
"server" -> {
val host = child.renderAttribute("host") ?: throw ConfigurationException("host attribute is required")
val port = child.renderAttribute("port")?.toInt() ?: throw ConfigurationException("port attribute is required")
val maxConnections = child.renderAttribute("max-connections")?.toInt() ?: 1
val connectionTimeout = child.renderAttribute("connection-timeout")
?.let(Duration::parse)
?.let(Duration::toMillis)
?.let(Long::toInt)
?: 10000
servers.add(MemcacheCacheConfiguration.Server(HostAndPort(host, port), connectionTimeout, maxConnections))
}
}
}
return MemcacheCacheConfiguration(
servers,
maxAge,
maxSize,
digestAlgorithm,
compressionMode,
)
}
override fun serialize(doc: Document, cache: MemcacheCacheConfiguration) = cache.run {
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") {
attr("host", server.endpoint.host)
attr("port", server.endpoint.port.toString())
server.connectionTimeoutMillis?.let { connectionTimeoutMillis ->
attr("connection-timeout", Duration.of(connectionTimeoutMillis.toLong(), ChronoUnit.MILLIS).toString())
}
attr("max-connections", server.maxConnections.toString())
}
}
attr("max-age", maxAge.toString())
attr("max-size", maxSize.toString())
digestAlgorithm?.let { digestAlgorithm ->
attr("digest", digestAlgorithm)
}
compressionMode?.let { compressionMode ->
attr(
"compression-mode", when (compressionMode) {
MemcacheCacheConfiguration.CompressionMode.GZIP -> "gzip"
MemcacheCacheConfiguration.CompressionMode.DEFLATE -> "deflate"
}
)
}
}
result
}
}

View File

@@ -0,0 +1,257 @@
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
import io.netty.channel.ChannelPipeline
import io.netty.channel.SimpleChannelInboundHandler
import io.netty.channel.nio.NioEventLoopGroup
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.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.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.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 io.netty.util.concurrent.Future as NettyFuture
class MemcacheClient(private val cfg: MemcacheCacheConfiguration) : AutoCloseable {
private companion object {
@JvmStatic
private val log = contextLogger()
}
private val group: NioEventLoopGroup
private val connectionPool: MutableMap<HostAndPort, ChannelPool> = ConcurrentHashMap()
init {
group = NioEventLoopGroup()
}
private fun newConnectionPool(server: MemcacheCacheConfiguration.Server): FixedChannelPool {
val bootstrap = Bootstrap().apply {
group(group)
channel(NioSocketChannel::class.java)
option(ChannelOption.SO_KEEPALIVE, true)
remoteAddress(InetSocketAddress(server.endpoint.host, server.endpoint.port))
server.connectionTimeoutMillis?.let {
option(ChannelOption.CONNECT_TIMEOUT_MILLIS, it)
}
}
val channelPoolHandler = object : AbstractChannelPoolHandler() {
override fun channelCreated(ch: Channel) {
val pipeline: ChannelPipeline = ch.pipeline()
pipeline.addLast(BinaryMemcacheClientCodec())
pipeline.addLast(BinaryMemcacheObjectAggregator(cfg.maxSize))
}
}
return FixedChannelPool(bootstrap, channelPoolHandler, server.maxConnections)
}
private fun sendRequest(request: FullBinaryMemcacheRequest): CompletableFuture<FullBinaryMemcacheResponse> {
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()
checksum = checksum xor byte
}
while (key.readableBytes() > 0) {
val byte = key.readByte()
checksum = checksum xor byte.toInt()
}
servers[checksum % servers.size]
} else {
servers.first()
}
}
val response = CompletableFuture<FullBinaryMemcacheResponse>()
// Custom handler for processing responses
val pool = connectionPool.computeIfAbsent(server.endpoint) {
newConnectionPool(server)
}
pool.acquire().addListener(object : GenericFutureListener<NettyFuture<Channel>> {
override fun operationComplete(channelFuture: NettyFuture<Channel>) {
if (channelFuture.isSuccess) {
val channel = channelFuture.now
val pipeline = channel.pipeline()
channel.pipeline()
.addLast("client-handler", object : SimpleChannelInboundHandler<FullBinaryMemcacheResponse>() {
override fun channelRead0(
ctx: ChannelHandlerContext,
msg: FullBinaryMemcacheResponse
) {
pipeline.removeLast()
pool.release(channel)
msg.touch("The method's caller must remember to release this")
response.complete(msg.retain())
}
override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
val ex = when (cause) {
is DecoderException -> cause.cause!!
else -> cause
}
ctx.close()
pipeline.removeLast()
pool.release(channel)
response.completeExceptionally(ex)
}
})
request.touch()
channel.writeAndFlush(request)
} else {
response.completeExceptionally(channelFuture.cause())
}
}
})
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()
}
override fun close() {
shutDown().sync()
}
}

View File

@@ -0,0 +1 @@
net.woggioni.rbcs.server.memcache.MemcacheCacheProvider

View File

@@ -1,33 +1,35 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<xs:schema targetNamespace="urn:net.woggioni.gbcs.server.memcached"
xmlns:gbcs-memcached="urn:net.woggioni.gbcs.server.memcached"
xmlns:gbcs="urn:net.woggioni.gbcs.server"
<xs:schema targetNamespace="urn:net.woggioni.rbcs.server.memcache"
xmlns:rbcs-memcache="urn:net.woggioni.rbcs.server.memcache"
xmlns:rbcs="urn:net.woggioni.rbcs.server"
xmlns:xs="http://www.w3.org/2001/XMLSchema">
<xs:import schemaLocation="jpms://net.woggioni.gbcs.server/net/woggioni/gbcs/server/schema/gbcs.xsd" namespace="urn:net.woggioni.gbcs.server"/>
<xs:import schemaLocation="jpms://net.woggioni.rbcs.server/net/woggioni/rbcs/server/schema/rbcs.xsd" namespace="urn:net.woggioni.rbcs.server"/>
<xs:complexType name="memcachedServerType">
<xs:complexType name="memcacheServerType">
<xs:attribute name="host" type="xs:token" use="required"/>
<xs:attribute name="port" type="xs:positiveInteger" use="required"/>
<xs:attribute name="connection-timeout" type="xs:duration"/>
<xs:attribute name="max-connections" type="xs:positiveInteger" default="1"/>
</xs:complexType>
<xs:complexType name="memcachedCacheType">
<xs:complexType name="memcacheCacheType">
<xs:complexContent>
<xs:extension base="gbcs:cacheType">
<xs:extension base="rbcs:cacheType">
<xs:sequence maxOccurs="unbounded">
<xs:element name="server" type="gbcs-memcached:memcachedServerType"/>
<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="digest" type="xs:token" />
<xs:attribute name="compression-mode" type="gbcs-memcached:compressionType" default="zip"/>
<xs:attribute name="compression-mode" type="rbcs-memcache:compressionType"/>
</xs:extension>
</xs:complexContent>
</xs:complexType>
<xs:simpleType name="compressionType">
<xs:restriction base="xs:token">
<xs:enumeration value="zip"/>
<xs:enumeration value="deflate"/>
<xs:enumeration value="gzip"/>
</xs:restriction>
</xs:simpleType>

View File

@@ -10,19 +10,16 @@ dependencies {
implementation catalog.slf4j.api
implementation catalog.netty.codec.http
api project(':gbcs-common')
api project(':gbcs-api')
api project(':rbcs-common')
api project(':rbcs-api')
// runtimeOnly catalog.slf4j.jdk14
testRuntimeOnly catalog.logback.classic
testImplementation catalog.bcprov.jdk18on
testImplementation catalog.bcpkix.jdk18on
testImplementation catalog.junit.jupiter.api
testImplementation catalog.junit.jupiter.params
testRuntimeOnly catalog.junit.jupiter.engine
testRuntimeOnly project(":gbcs-server-memcached")
testRuntimeOnly project(":rbcs-server-memcache")
}
test {

View File

@@ -1,8 +1,8 @@
import net.woggioni.gbcs.api.CacheProvider;
import net.woggioni.gbcs.server.cache.FileSystemCacheProvider;
import net.woggioni.gbcs.server.cache.InMemoryCacheProvider;
import net.woggioni.rbcs.api.CacheProvider;
import net.woggioni.rbcs.server.cache.FileSystemCacheProvider;
import net.woggioni.rbcs.server.cache.InMemoryCacheProvider;
module net.woggioni.gbcs.server {
module net.woggioni.rbcs.server {
requires java.sql;
requires java.xml;
requires java.logging;
@@ -16,13 +16,13 @@ module net.woggioni.gbcs.server {
requires io.netty.codec;
requires org.slf4j;
requires net.woggioni.jwo;
requires net.woggioni.gbcs.common;
requires net.woggioni.gbcs.api;
requires net.woggioni.rbcs.common;
requires net.woggioni.rbcs.api;
exports net.woggioni.gbcs.server;
exports net.woggioni.rbcs.server;
opens net.woggioni.gbcs.server;
opens net.woggioni.gbcs.server.schema;
opens net.woggioni.rbcs.server;
opens net.woggioni.rbcs.server.schema;
uses CacheProvider;
provides CacheProvider with FileSystemCacheProvider, InMemoryCacheProvider;

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.server
package net.woggioni.rbcs.server
import io.netty.channel.ChannelHandlerContext
import org.slf4j.Logger

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.server
package net.woggioni.rbcs.server
import io.netty.bootstrap.ServerBootstrap
import io.netty.buffer.ByteBuf
@@ -30,24 +30,24 @@ 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.gbcs.api.Configuration
import net.woggioni.gbcs.api.exception.ConfigurationException
import net.woggioni.gbcs.common.GBCS.toUrl
import net.woggioni.gbcs.common.PasswordSecurity.decodePasswordHash
import net.woggioni.gbcs.common.PasswordSecurity.hashPassword
import net.woggioni.gbcs.common.Xml
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.common.debug
import net.woggioni.gbcs.common.info
import net.woggioni.gbcs.server.auth.AbstractNettyHttpAuthenticator
import net.woggioni.gbcs.server.auth.Authorizer
import net.woggioni.gbcs.server.auth.ClientCertificateValidator
import net.woggioni.gbcs.server.auth.RoleAuthorizer
import net.woggioni.gbcs.server.configuration.Parser
import net.woggioni.gbcs.server.configuration.Serializer
import net.woggioni.gbcs.server.exception.ExceptionHandler
import net.woggioni.gbcs.server.handler.ServerHandler
import net.woggioni.gbcs.server.throttling.ThrottlingHandler
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.Xml
import net.woggioni.rbcs.common.contextLogger
import net.woggioni.rbcs.common.debug
import net.woggioni.rbcs.common.info
import net.woggioni.rbcs.server.auth.AbstractNettyHttpAuthenticator
import net.woggioni.rbcs.server.auth.Authorizer
import net.woggioni.rbcs.server.auth.ClientCertificateValidator
import net.woggioni.rbcs.server.auth.RoleAuthorizer
import net.woggioni.rbcs.server.configuration.Parser
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
@@ -65,7 +65,7 @@ import java.util.regex.Pattern
import javax.naming.ldap.LdapName
import javax.net.ssl.SSLPeerUnverifiedException
class GradleBuildCacheServer(private val cfg: Configuration) {
class RemoteBuildCacheServer(private val cfg: Configuration) {
private val log = contextLogger()
companion object {
@@ -73,7 +73,7 @@ class GradleBuildCacheServer(private val cfg: Configuration) {
val userAttribute: AttributeKey<Configuration.User> = AttributeKey.valueOf("user")
val groupAttribute: AttributeKey<Set<Configuration.Group>> = AttributeKey.valueOf("group")
val DEFAULT_CONFIGURATION_URL by lazy { "classpath:net/woggioni/gbcs/gbcs-default.xml".toUrl() }
val DEFAULT_CONFIGURATION_URL by lazy { "classpath:net/woggioni/rbcs/server/rbcs-default.xml".toUrl() }
private const val SSL_HANDLER_NAME = "sslHandler"
fun loadConfiguration(configurationFile: Path): Configuration {
@@ -208,15 +208,15 @@ class GradleBuildCacheServer(private val cfg: Configuration) {
.map { it as X509Certificate }
.toArray { size -> Array<X509Certificate?>(size) { null } }
SslContextBuilder.forServer(serverKey, *serverCert).apply {
if (tls.isVerifyClients) {
clientAuth(ClientAuth.OPTIONAL)
tls.trustStore?.let { trustStore ->
val clientAuth = tls.trustStore?.let { trustStore ->
val ts = loadKeystore(trustStore.file, trustStore.password)
trustManager(
ClientCertificateValidator.getTrustManager(ts, trustStore.isCheckCertificateStatus)
)
}
}
if(trustStore.isRequireClientCertificate) ClientAuth.REQUIRE
else ClientAuth.OPTIONAL
} ?: ClientAuth.NONE
clientAuth(clientAuth)
}.build()
}
}
@@ -393,16 +393,16 @@ class GradleBuildCacheServer(private val cfg: Configuration) {
}
}
log.info {
"GradleBuildCacheServer has been gracefully shut down"
"RemoteBuildCacheServer has been gracefully shut down"
}
}
}
fun run(): ServerHandle {
// Create the multithreaded event loops for the server
val bossGroup = NioEventLoopGroup(0)
val bossGroup = NioEventLoopGroup(1)
val serverSocketChannel = NioServerSocketChannel::class.java
val workerGroup = bossGroup
val workerGroup = NioEventLoopGroup(0)
val eventExecutorGroup = run {
val threadFactory = if (cfg.eventExecutor.isUseVirtualThreads) {
Thread.ofVirtual().factory()
@@ -425,7 +425,7 @@ class GradleBuildCacheServer(private val cfg: Configuration) {
val bindAddress = InetSocketAddress(cfg.host, cfg.port)
val httpChannel = bootstrap.bind(bindAddress).sync()
log.info {
"GradleBuildCacheServer is listening on ${cfg.host}:${cfg.port}"
"RemoteBuildCacheServer is listening on ${cfg.host}:${cfg.port}"
}
return ServerHandle(httpChannel, setOf(bossGroup, workerGroup, eventExecutorGroup))
}

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.server.auth
package net.woggioni.rbcs.server.auth
import io.netty.buffer.Unpooled
import io.netty.channel.ChannelFutureListener
@@ -11,10 +11,10 @@ import io.netty.handler.codec.http.HttpRequest
import io.netty.handler.codec.http.HttpResponseStatus
import io.netty.handler.codec.http.HttpVersion
import io.netty.util.ReferenceCountUtil
import net.woggioni.gbcs.api.Configuration
import net.woggioni.gbcs.api.Configuration.Group
import net.woggioni.gbcs.api.Role
import net.woggioni.gbcs.server.GradleBuildCacheServer
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.api.Configuration.Group
import net.woggioni.rbcs.api.Role
import net.woggioni.rbcs.server.RemoteBuildCacheServer
abstract class AbstractNettyHttpAuthenticator(private val authorizer: Authorizer) : ChannelInboundHandlerAdapter() {
@@ -40,8 +40,8 @@ abstract class AbstractNettyHttpAuthenticator(private val authorizer: Authorizer
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
if (msg is HttpRequest) {
val result = authenticate(ctx, msg) ?: return authenticationFailure(ctx, msg)
ctx.channel().attr(GradleBuildCacheServer.userAttribute).set(result.user)
ctx.channel().attr(GradleBuildCacheServer.groupAttribute).set(result.groups)
ctx.channel().attr(RemoteBuildCacheServer.userAttribute).set(result.user)
ctx.channel().attr(RemoteBuildCacheServer.groupAttribute).set(result.groups)
val roles = (
(result.user?.let { user ->

View File

@@ -1,7 +1,7 @@
package net.woggioni.gbcs.server.auth
package net.woggioni.rbcs.server.auth
import io.netty.handler.codec.http.HttpRequest
import net.woggioni.gbcs.api.Role
import net.woggioni.rbcs.api.Role
fun interface Authorizer {
fun authorize(roles : Set<Role>, request: HttpRequest) : Boolean

View File

@@ -1,4 +1,4 @@
package net.woggioni.gbcs.server.auth
package net.woggioni.rbcs.server.auth
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter

View File

@@ -1,8 +1,8 @@
package net.woggioni.gbcs.server.auth
package net.woggioni.rbcs.server.auth
import io.netty.handler.codec.http.HttpMethod
import io.netty.handler.codec.http.HttpRequest
import net.woggioni.gbcs.api.Role
import net.woggioni.rbcs.api.Role
class RoleAuthorizer : Authorizer {

View File

@@ -1,8 +1,11 @@
package net.woggioni.gbcs.server.cache
package net.woggioni.rbcs.server.cache
import net.woggioni.gbcs.api.Cache
import net.woggioni.gbcs.common.contextLogger
import net.woggioni.gbcs.server.cache.CacheUtils.digestString
import io.netty.buffer.ByteBuf
import net.woggioni.rbcs.api.Cache
import net.woggioni.rbcs.common.ByteBufInputStream
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 java.nio.channels.FileChannel
@@ -14,6 +17,7 @@ import java.nio.file.attribute.BasicFileAttributes
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
@@ -28,7 +32,10 @@ class FileSystemCache(
val compressionLevel: Int
) : Cache {
private companion object {
@JvmStatic
private val log = contextLogger()
}
init {
Files.createDirectories(root)
@@ -62,10 +69,12 @@ class FileSystemCache(
}
}.also {
gc()
}.let {
CompletableFuture.completedFuture(it)
}
}
override fun put(key: String, content: ByteArray) {
override fun put(key: String, content: ByteBuf): CompletableFuture<Void> {
(digestAlgorithm
?.let(MessageDigest::getInstance)
?.let { md ->
@@ -82,7 +91,7 @@ class FileSystemCache(
it
}
}.use {
it.write(content)
JWO.copy(ByteBufInputStream(content), it)
}
Files.move(tmpFile, file, StandardCopyOption.ATOMIC_MOVE)
} catch (t: Throwable) {
@@ -92,6 +101,7 @@ class FileSystemCache(
}.also {
gc()
}
return CompletableFuture.completedFuture(null)
}
private fun gc() {

View File

@@ -1,7 +1,7 @@
package net.woggioni.gbcs.server.cache
package net.woggioni.rbcs.server.cache
import net.woggioni.gbcs.api.Configuration
import net.woggioni.gbcs.common.GBCS
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.common.RBCS
import net.woggioni.jwo.Application
import java.nio.file.Path
import java.time.Duration
@@ -14,14 +14,14 @@ data class FileSystemCacheConfiguration(
val compressionLevel: Int,
) : Configuration.Cache {
override fun materialize() = FileSystemCache(
root ?: Application.builder("gbcs").build().computeCacheDirectory(),
root ?: Application.builder("rbcs").build().computeCacheDirectory(),
maxAge,
digestAlgorithm,
compressionEnabled,
compressionLevel
)
override fun getNamespaceURI() = GBCS.GBCS_NAMESPACE_URI
override fun getNamespaceURI() = RBCS.RBCS_NAMESPACE_URI
override fun getTypeName() = "fileSystemCacheType"
}

View File

@@ -1,9 +1,9 @@
package net.woggioni.gbcs.server.cache
package net.woggioni.rbcs.server.cache
import net.woggioni.gbcs.api.CacheProvider
import net.woggioni.gbcs.common.GBCS
import net.woggioni.gbcs.common.Xml
import net.woggioni.gbcs.common.Xml.Companion.renderAttribute
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.common.RBCS
import net.woggioni.rbcs.common.Xml
import net.woggioni.rbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import org.w3c.dom.Element
import java.nio.file.Path
@@ -12,11 +12,11 @@ import java.util.zip.Deflater
class FileSystemCacheProvider : CacheProvider<FileSystemCacheConfiguration> {
override fun getXmlSchemaLocation() = "classpath:net/woggioni/gbcs/server/schema/gbcs.xsd"
override fun getXmlSchemaLocation() = "classpath:net/woggioni/rbcs/server/schema/rbcs.xsd"
override fun getXmlType() = "fileSystemCacheType"
override fun getXmlNamespace() = "urn:net.woggioni.gbcs.server"
override fun getXmlNamespace() = "urn:net.woggioni.rbcs.server"
override fun deserialize(el: Element): FileSystemCacheConfiguration {
val path = el.renderAttribute("path")
@@ -44,8 +44,8 @@ class FileSystemCacheProvider : CacheProvider<FileSystemCacheConfiguration> {
override fun serialize(doc: Document, cache : FileSystemCacheConfiguration) = cache.run {
val result = doc.createElement("cache")
Xml.of(doc, result) {
val prefix = doc.lookupPrefix(GBCS.GBCS_NAMESPACE_URI)
attr("xs:type", "${prefix}:fileSystemCacheType", GBCS.XML_SCHEMA_NAMESPACE_URI)
val prefix = doc.lookupPrefix(RBCS.RBCS_NAMESPACE_URI)
attr("xs:type", "${prefix}:fileSystemCacheType", RBCS.XML_SCHEMA_NAMESPACE_URI)
attr("path", root.toString())
attr("max-age", maxAge.toString())
digestAlgorithm?.let { digestAlgorithm ->

View File

@@ -0,0 +1,150 @@
package net.woggioni.rbcs.server.cache
import io.netty.buffer.ByteBuf
import net.woggioni.rbcs.api.Cache
import net.woggioni.rbcs.common.ByteBufInputStream
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 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.atomic.AtomicLong
import java.util.zip.Deflater
import java.util.zip.DeflaterOutputStream
import java.util.zip.Inflater
import java.util.zip.InflaterInputStream
class InMemoryCache(
val maxAge: Duration,
val maxSize: Long,
val digestAlgorithm: String?,
val compressionEnabled: Boolean,
val compressionLevel: Int
) : Cache {
companion object {
@JvmStatic
private val log = contextLogger()
}
private val size = AtomicLong()
private val map = ConcurrentHashMap<String, ByteBuf>()
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>()
private var running = true
private val garbageCollector = Thread {
while(true) {
val el = removalQueue.take()
val buf = el.value
val now = Instant.now()
if(now > el.expiry) {
val removed = map.remove(el.key, buf)
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) {
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) {
val newSize = updateSizeAfterRemoval(buf)
//Decrease the reference count for map
buf.release()
return newSize
}
}
}
private fun updateSizeAfterRemoval(removed: ByteBuf) : Long {
return size.updateAndGet { currentSize : Long ->
currentSize - removed.readableBytes()
}
}
override fun close() {
running = false
garbageCollector.join()
}
override fun get(key: String) =
(digestAlgorithm
?.let(MessageDigest::getInstance)
?.let { md ->
digestString(key.toByteArray(), md)
} ?: key
).let { digest ->
map[digest]
?.let { value ->
val copy = value.retainedDuplicate()
copy.touch("This has to be released by the caller of the cache")
if (compressionEnabled) {
val inflater = Inflater()
Channels.newChannel(InflaterInputStream(ByteBufInputStream(copy), inflater))
} else {
Channels.newChannel(ByteBufInputStream(copy))
}
}
}.let {
CompletableFuture.completedFuture(it)
}
override fun put(key: String, content: ByteBuf) =
(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)
var newSize = size.updateAndGet { currentSize : Long ->
currentSize + delta
}
removalQueue.put(RemovalQueueElement(digest, value.retain(), Instant.now().plus(maxAge)))
while(newSize > maxSize) {
newSize = removeEldest()
}
}.let {
CompletableFuture.completedFuture<Void>(null)
}
}

View File

@@ -1,23 +1,25 @@
package net.woggioni.gbcs.server.cache
package net.woggioni.rbcs.server.cache
import net.woggioni.gbcs.api.Configuration
import net.woggioni.gbcs.common.GBCS
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.common.RBCS
import java.time.Duration
data class InMemoryCacheConfiguration(
val maxAge: Duration,
val maxSize: Long,
val digestAlgorithm : String?,
val compressionEnabled: Boolean,
val compressionLevel: Int,
) : Configuration.Cache {
override fun materialize() = InMemoryCache(
maxAge,
maxSize,
digestAlgorithm,
compressionEnabled,
compressionLevel
)
override fun getNamespaceURI() = GBCS.GBCS_NAMESPACE_URI
override fun getNamespaceURI() = RBCS.RBCS_NAMESPACE_URI
override fun getTypeName() = "inMemoryCacheType"
}

View File

@@ -1,27 +1,29 @@
package net.woggioni.gbcs.server.cache
package net.woggioni.rbcs.server.cache
import net.woggioni.gbcs.api.CacheProvider
import net.woggioni.gbcs.common.GBCS
import net.woggioni.gbcs.common.Xml
import net.woggioni.gbcs.common.Xml.Companion.renderAttribute
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.common.RBCS
import net.woggioni.rbcs.common.Xml
import net.woggioni.rbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import org.w3c.dom.Element
import java.nio.file.Path
import java.time.Duration
import java.util.zip.Deflater
class InMemoryCacheProvider : CacheProvider<InMemoryCacheConfiguration> {
override fun getXmlSchemaLocation() = "classpath:net/woggioni/gbcs/server/schema/gbcs.xsd"
override fun getXmlSchemaLocation() = "classpath:net/woggioni/rbcs/server/schema/rbcs.xsd"
override fun getXmlType() = "inMemoryCacheType"
override fun getXmlNamespace() = "urn:net.woggioni.gbcs.server"
override fun getXmlNamespace() = "urn:net.woggioni.rbcs.server"
override fun deserialize(el: Element): InMemoryCacheConfiguration {
val maxAge = el.renderAttribute("max-age")
?.let(Duration::parse)
?: Duration.ofDays(1)
val maxSize = el.renderAttribute("max-size")
?.let(java.lang.Long::decode)
?: 0x1000000
val enableCompression = el.renderAttribute("enable-compression")
?.let(String::toBoolean)
?: true
@@ -32,6 +34,7 @@ class InMemoryCacheProvider : CacheProvider<InMemoryCacheConfiguration> {
return InMemoryCacheConfiguration(
maxAge,
maxSize,
digestAlgorithm,
enableCompression,
compressionLevel
@@ -41,9 +44,10 @@ class InMemoryCacheProvider : CacheProvider<InMemoryCacheConfiguration> {
override fun serialize(doc: Document, cache : InMemoryCacheConfiguration) = cache.run {
val result = doc.createElement("cache")
Xml.of(doc, result) {
val prefix = doc.lookupPrefix(GBCS.GBCS_NAMESPACE_URI)
attr("xs:type", "${prefix}:inMemoryCacheType", GBCS.XML_SCHEMA_NAMESPACE_URI)
val prefix = doc.lookupPrefix(RBCS.RBCS_NAMESPACE_URI)
attr("xs:type", "${prefix}:inMemoryCacheType", RBCS.XML_SCHEMA_NAMESPACE_URI)
attr("max-age", maxAge.toString())
attr("max-size", maxSize.toString())
digestAlgorithm?.let { digestAlgorithm ->
attr("digest", digestAlgorithm)
}

View File

@@ -1,7 +1,7 @@
package net.woggioni.gbcs.server.configuration
package net.woggioni.rbcs.server.configuration
import net.woggioni.gbcs.api.CacheProvider
import net.woggioni.gbcs.api.Configuration
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.api.Configuration
import java.util.ServiceLoader
object CacheSerializers {

Some files were not shown because too many files have changed in this diff Show More