Skip to content
BranchesTags

Repository files navigation

Open Application Platform

A light-weight application framework to build high performant and distributed java applications.

Modules

Module Description
oap-application IoC/DI Kernel — discovers services from HOCON descriptors, wires dependencies, manages start/stop lifecycle
oap-stdlib Core utilities — Binder (JSON/HOCON/YAML/XML), Files, IoStreams, Cuid, Dates, Stream, Result
oap-http Undertow-based HTTP server with named ports, PNIO high-performance pipeline, and HTTP client
oap-ws Annotation-driven web services (@WsMethod, @WsParam) with session and interceptor support
oap-jpath JPath expression language for navigating object graphs: ${var.field.method().array[n]}
oap-formats Template engine, TSV/CSV parsing, JSON schema validation, and log streaming
oap-statsdb Distributed in-memory statistics tree with hierarchical rollup and MongoDB persistence
oap-message Reliable binary message delivery with disk spill, retry, and MD5-based deduplication
oap-storage In-memory object store (MemoryStorage) with MongoDB sync and cloud object storage
oap-highload CPU affinity utility for pinning threads to specific CPU cores
oap-mail Email sending via SMTP and SendGrid with a persistent delivery queue
oap-notification Pub/sub notification delivery via a pluggable NotificationTransport (MQTT provided)
oap-maven-plugin Build-time code generation: startup scripts and dictionary enum source files

Guides

Guide Description
Ext — Pluggable Field Extensions Attach pluggable typed sub-objects to bean fields via oap.json.ext.Ext; covers JSON deserialization and template engine integration
Testing Fixture lifecycle, KernelFixture, assertion helpers, MongoDB/S3 mocks, and benchmark harness

Related Projects

Project Description
oap-config HOCON-based configuration library used by the Kernel to parse oap-module.oap / application.conf
oap-config-plugin IntelliJ IDEA plugin: HOCON language support (syntax highlighting, references)
oap-application-plugin IntelliJ IDEA plugin: navigation/completion for OAP Kernel service definitions

oap-application

The IoC/DI kernel for the OAP framework. Kernel discovers service descriptors from every jar on the classpath, builds a dependency graph, instantiates and wires services, and manages their full lifecycle (start, scheduled runs, stop).

Services are plain Java classes — no framework annotations required. Everything is declared in HOCON files.

Table of Contents

Overview

Startup sequence:

1. Scan classpath for all META-INF/oap-module.oap files
2. Load and validate each module descriptor
3. Read application.conf (+ conf.d/ + CONFIG.* env vars)
4. Select modules reachable via boot.main transitive dependsOn graph
5. Topological sort modules and services
6. Instantiate each service (constructor or field injection)
7. Wire references, listeners, and links
8. Start supervised services via Supervisor

Boot is the production entry point. KernelFixture is the test entry point.

Module Declaration (oap-module.oap)

Every OAP jar ships a descriptor at src/main/resources/META-INF/oap-module.oap.

Module fields

Field Type Default Meaning
name String required Unique module identifier; must match [A-Za-z\-_0-9]+
enabled boolean true Disable the entire module and all its services
dependsOn list<String> [] Module-level ordering: this module starts after listed modules
services map required Map of service name → service block

Service fields

Field Type Default Meaning
implementation String required Fully-qualified class name
abstract boolean false Marks an interface/abstract-class slot; must be filled at deployment via application.conf
default reference — Default concrete implementation used when abstract=true and nothing is specified in application.conf
enabled boolean true Disable this service individually
parameters map {} Constructor or field values; may contain <...> references
supervision block — Lifecycle management (see Supervision)
dependsOn list [] Explicit start-order hints within a module
listen map {} Listener registration: listenerName = <ref> calls ref.addListenerNameListener(this)
link map {} Reverse wiring: fieldName = <ref> calls ref.addFieldName(this) / ref.setFieldName(this) / appends to collection

The service and services keys are aliases for the services map.

Example: two-module setup

m1.oap

name = m1

services {
  cm {
    implementation = com.example.ComplexMap
  }

  ServiceOneP1 {
    implementation = com.example.ServiceOne
    parameters {
      i          = 2ms
      kernel     = <kernel.self>
      complexMap = <modules.this.cm>
      complex {
        i = 2
        map.a.i = 1
      }
      complexes = [{i = 2}]
    }
    supervision.delay = 5ms
  }
}

m2.oap

name = m2
dependsOn = m1

services {
  ServiceTwo {
    implementation = com.example.ServiceTwo
    parameters {
      j   = 1
      one = <modules.m1.ServiceOneP1>
    }
    listen.some = <modules.m1.ServiceOneP1>
    supervision.supervise = true
  }

  ServiceScheduled {
    implementation = com.example.ServiceScheduled
    supervision {
      schedule = true
      delay    = 1s
    }
  }
}

m3.oap (list parameter referencing services from two modules)

name = m3
dependsOn = [m1, m2]

services {
  ServiceDepsList {
    implementation = com.example.ServiceDepsList
    parameters {
      deps = [
        <modules.m1.ServiceOneP1>
        <modules.m2.ServiceTwo>
      ]
    }
  }
}

Application Configuration (application.conf)

application.conf selects which modules activate and overrides their parameters at deployment time.

Schema

boot.main = [m1, m2, m3]   # one or more module names

shutdown {
  serviceTimeout                = 5s      # warn timeout per service during stop
  serviceAsyncShutdownAfterTimeout = false  # continue stopping if timeout exceeded
}

services {
  # Override a parameter
  m1.ServiceOneP1.parameters.i = 100ms

  # Disable a service
  m2.ServiceTwo.enabled = false

  # Assign a concrete implementation to an abstract service slot
  my-module.my-abstract-service = <modules.impl-module.ConcreteImpl>
}

conf.d/ directory

All *.conf and *.yaml files in conf.d/ are merged with application.conf. The default conf.d path is <application.conf parent>/conf.d. Useful for splitting deployment-specific values across files.

# conf.d/ports.conf
services.my-module.my-service.parameters.port = 9090

# conf.d/db.yaml
services:
  my-module:
    db-service:
      parameters:
        url: jdbc:postgresql://localhost/mydb

CONFIG.* environment variable overrides

Any environment variable starting with CONFIG. is stripped of the prefix and injected as a HOCON key. This allows per-deployment overrides without modifying config files.

export CONFIG.services.my-module.my-service.enabled=false
export CONFIG.services.my-module.my-service.parameters.val='"hello"'

Programmatic startup (tests / embedded)

kernel.start( Map.of(
    "boot.main", "m1",
    "services.m1.ServiceOneP1.parameters.i", "50"
) );

Reference Syntax

<...> expressions in parameter values are resolved by the kernel before service construction.

Expression Resolves to
<modules.moduleName.serviceName> The live instance of the named service
<modules.this.serviceName> A service in the same module
<modules.self.serviceName> Alias for this
<modules.*.serviceName> First matching service across all modules
<kernel.self> The Kernel instance itself
<services.self.name> The string name of the current service
location.module The URL of the module's own .oap file

References work in parameters, list parameters, and map parameters.

# Inject the kernel itself
parameters.kernel = <kernel.self>

# Cross-module reference
parameters.server = <modules.oap-http.oap-http-server>

# Same-module reference
parameters.cache = <modules.this.cache-service>

# Wildcard: first service named "config" in any module
parameters.config = <modules.*.config>

# Service's own registered name
parameters.serviceName = <services.self.name>

Supervision and Service Lifecycle

The supervision block controls how the kernel starts, runs, and stops a service.

Fields

Field Type Default Effect
supervise boolean false Call lifecycle methods on start/stop
thread boolean false Run service as a Runnable in a dedicated daemon thread
schedule boolean false Run service periodically (combine with delay or cron)
delay duration 0 Fixed-delay interval; supports HOCON duration units (1s, 5ms, 1h)
cron String — Quartz cron expression for scheduled runs
preStartWith list<String> ["preStart"] Method names called before start
startWith list<String> ["start"] Method names called on start
preStopWith list<String> ["preStop"] Method names called before stop
stopWith list<String> ["stop","close"] Method names called on stop; services implementing Closeable get close() called automatically

Missing lifecycle methods are silently skipped.

Lifecycle examples

Supervised service — lifecycle methods called in order:

service {
  implementation = com.example.MyService   # has preStart/start/preStop/stop methods
  supervision.supervise = true
}

Result on start: preStart() → start()
Result on stop: preStop() → stop()

Supervised thread — runs Runnable.run() in a daemon thread:

thread {
  implementation = com.example.WorkerService
  supervision {
    supervise = true
    thread    = true
  }
}

Delay-scheduled — runs Runnable.run() every N ms with fixed delay:

poller {
  implementation = com.example.PollerService
  supervision {
    schedule = true
    delay    = 30s
  }
}

Cron-scheduled — runs Runnable.run() on a Quartz cron schedule:

nightly-cleanup {
  implementation = com.example.CleanupJob
  supervision {
    supervise = true
    schedule  = true
    cron      = "0 0 2 * * ? *"   # every day at 02:00 UTC
  }
}

When supervise = true is combined with schedule = true, lifecycle methods are called around the entire scheduler lifetime (start before first run, stop after last run).

Shutdown sequence

On kernel.stop(), the Supervisor stops services in reverse registration order:

  1. Threads and scheduled tasks are interrupted/cancelled.
  2. Supervised services have preStop() then stop() (or close()) called.

shutdown.serviceTimeout (default 5s) is the warn threshold per service. Set shutdown.serviceAsyncShutdownAfterTimeout = true to continue shutdown after a timeout rather than waiting.

Lifecycle Annotations

As an alternative to relying on method names (preStart, start, etc.), lifecycle hooks can be declared with annotations from the oap.application.annotation package. The kernel discovers annotated methods at startup regardless of their name.

Annotation Phase Equivalent supervision field
@PreStart Before service start preStartWith
@Start Service start startWith
@PreStop Before service stop preStopWith
@Stop Service stop stopWith

Annotations and name-based discovery are independent — both are applied. An annotated method named start is invoked by both paths; an annotated method with any other name is invoked only via the annotation.

import oap.application.annotation.PreStart;
import oap.application.annotation.Start;
import oap.application.annotation.PreStop;
import oap.application.annotation.Stop;

public class MyService {
    @PreStart
    public void onBeforeStart() { /* runs before start */ }

    @Start
    public void onStart() { /* runs on start */ }

    @PreStop
    public void onBeforeStop() { /* runs before stop */ }

    @Stop
    public void onStop() { /* runs on stop */ }
}

The service still needs supervision.supervise = true in its oap-module.oap declaration.

Dependency Injection Mechanics

Constructor injection

The kernel reflects on the service class and calls a constructor whose parameter names match keys in parameters. References are resolved first, then scalars are type-coerced.

public class ServiceTwo {
    public ServiceTwo( ServiceOne one, int j ) { ... }
}
ServiceTwo {
  implementation = com.example.ServiceTwo
  parameters {
    one = <modules.m1.ServiceOneP1>
    j   = 42
  }
}

Field injection

Parameters not consumed by the constructor are applied to public fields by name.

Nested object parameters

Maps in parameters are bound to nested objects:

parameters.complex {
  i = 2
  map.a.i = 1
}

Listen wiring

listen.name = <ref> — after construction, calls ref.addNameListener(this). The target service must have a method addNameListener(T listener).

ServiceTwo {
  listen.some = <modules.m1.ServiceOneP1>   # calls ServiceOneP1.addSomeListener(serviceTwo)
}

Link wiring (reverse injection)

link.name = <ref> — after construction, registers this on the target service. The kernel attempts in order:

  1. ref.addName(this)
  2. ref.setName(this)
  3. ref.addNameListener(this)
  4. ref.name field (appends if collection, sets otherwise)
ti1 {
  implementation = com.example.Impl
  link.registry = <modules.this.service-registry>
}

Disabled services

A disabled service referenced via <modules...> resolves to null and is omitted from list parameters. No error is thrown.

Cyclic dependencies

Cyclic module or service dependencies are detected at startup and throw ApplicationException("cyclic dependency detected").

Abstract Services

The abstract service pattern defines an interface slot in a module that must be filled with a concrete implementation — either by a default fallback or by an explicit assignment in application.conf.

Declaring an abstract service

# oap-module.oap
name = my-module

services {
  abstract-service {
    abstract        = true
    implementation  = com.example.AbstractService   # interface or abstract class
    default         = <modules.my-module.default-impl>  # optional fallback
  }

  service {
    implementation = com.example.Container
    parameters {
      dep        = <modules.this.abstract-service>
      fieldParam = <modules.this.abstract-service>
      listParam  = [<modules.this.abstract-service>]
    }
  }

  default-impl {
    implementation = com.example.DefaultImpl
  }
}

Selecting an implementation in application.conf

# application.conf
boot.main = my-module

services {
  my-module.abstract-service = <modules.my-module.default-impl>
}

Test-time mock

Create a test module that dependsOn the production module, disable the default implementation, and point the abstract service at the mock:

# test-module.oap
name = my-module-test
dependsOn = my-module

services {
  mock {
    implementation = com.example.MockImpl
  }
}
# application-test.conf
boot.main = my-module-test

services {
  my-module {
    default-impl.enabled   = false
    abstract-service       = <modules.my-module-test.mock>
  }
}

Error cases

  • abstract = true not set but implementation is an interface → ApplicationException: "abstract = true" property is missing
  • No concrete implementation registered and no default → ApplicationException: No implementation has been declared
  • Implementations exist but none selected → ApplicationException: No implementation specified ... Available implementations [...]

Module Discovery

Module.CONFIGURATION.urlsFromClassPath() scans all jars on the classpath for module descriptors in priority order:

  1. META-INF/oap-module.oap
  2. META-INF/oap-module.conf
  3. META-INF/oap-module.yaml / .yml
  4. META-INF/oap-module.json

All discovered modules are loaded. Only modules reachable via boot.main's transitive dependsOn graph are activated.

Module and service names must match the pattern ^[A-Za-z\-_0-9]++$.

boot.main transitive activation

With boot.main = [m1] and the graph m1 → m3 → m4, modules m1, m3, and m4 are activated. Module m2 (if unreachable from m1) is silently ignored.

oap-module-ext.conf

Attach typed metadata to service declarations by registering an extension type:

# META-INF/oap-module-ext.conf
services.ws.implementation = com.example.WsServiceExt

Then use the extension key freely in any module's service block (see KernelExt).

Kernel API

Construction

// Use all module descriptors from the classpath
Kernel kernel = new Kernel( Module.CONFIGURATION.urlsFromClassPath() );

// Named kernel (name appears in logs)
Kernel kernel = new Kernel( "my-app", Module.CONFIGURATION.urlsFromClassPath() );

Starting

kernel.start( Path.of( "/etc/myapp/application.conf" ) );
kernel.start( Path.of( "/etc/myapp/application.conf" ), Path.of( "/etc/myapp/conf.d" ) );
kernel.start( "classpath:application.conf", "conf.d" );
kernel.start( Map.of( "boot.main", "my-module" ) );   // programmatic (tests)
kernel.start( applicationConfiguration );              // pre-built config object

Service lookup

// Exact module + name
Optional<MyService> s = kernel.service( "my-module", "my-service" );

// By reference string
Optional<MyService> s = kernel.service( "my-module.my-service" );
Optional<MyService> s = kernel.service( "<modules.my-module.my-service>" );

// All services with a given name across all modules (use "*" for any module)
List<MyService> list = kernel.services( "*", "my-service" );

// By class
List<MyService>         all   = kernel.ofClass( MyService.class );
Optional<MyService>     first = kernel.serviceOfClass( MyService.class );
MyService               req   = kernel.serviceOfClass2( MyService.class );  // throws if not found

// Scoped to a module
List<MyService> list = kernel.ofClass( "my-module", MyService.class );

// By extension key (see KernelExt section)
List<ServiceExt<WsConfig>> wsServices = kernel.servicesByExt( "ws" );

Stopping

kernel.stop();

// Kernel implements Closeable — use try-with-resources in tests
try ( Kernel kernel = new Kernel( Module.CONFIGURATION.urlsFromClassPath() ) ) {
    kernel.start( Map.of( "boot.main", "my-module" ) );
    // assertions
}

KernelExt: Service Metadata Extensions

Arbitrary typed metadata can be attached to service declarations and queried at runtime. This is the mechanism used by oap-ws to discover HTTP-annotated services without scanning all services by type.

Registration

Add a line to META-INF/oap-module-ext.conf mapping the extension key to a Java class:

services.ws.implementation = com.example.WsServiceExt

WsServiceExt is a plain POJO that the HOCON binder will populate.

Usage in module file

services {
  my-api {
    implementation = com.example.MyApi
    ws {
      path = /api/v1
      port = httpprivate
    }
  }
}

Querying at runtime

List<ServiceExt<WsServiceExt>> endpoints = kernel.servicesByExt( "ws" );
for ( ServiceExt<WsServiceExt> ep : endpoints ) {
    System.out.println( ep.name + " → " + ep.ext.path );
}

Testing with KernelFixture

KernelFixture (from oap-application-test) is a TestNG fixture that starts a real Kernel before each test method and stops it after.

Variables automatically available in application.conf

Variable Value
TEST_HTTP_PORT A free HTTP port allocated for the test
TEST_DIRECTORY A per-test temp directory
TEST_RESOURCE_PATH Path to the test's resource directory
TEST_HTTP_PREFIX http://localhost:${TEST_HTTP_PORT}

Basic usage

@Listeners( Fixtures.class )
public class MyServiceTest {
    private final TestDirectoryFixture testDirectory = fixture( new TestDirectoryFixture() );
    private final KernelFixture kernel = fixture( new KernelFixture(
        testDirectory,
        Resources.url( MyServiceTest.class, "application.test.conf" ).orElseThrow(),
        List.of( Resources.url( MyServiceTest.class, "oap-module.oap" ).orElseThrow() )
    ) );

    @Test
    public void myTest() {
        MyService svc = kernel.service( "*", MyService.class ).orElseThrow();
        // test assertions
    }
}

application.test.conf pattern

boot.main = my-module

services {
  my-module {
    my-service.parameters.port = ${TEST_HTTP_PORT}
    my-service.parameters.dir  = ${TEST_DIRECTORY}
  }
}

Fluent builder methods

new KernelFixture( testDir, confUrl )
    .withProperties( Map.of( "MY_KEY", "value" ) )     // inject HOCON substitution vars
    .withConfResource( MyTest.class, "extra.conf" )     // merge extra conf file
    .withConfdResources( MyTest.class, "conf.d" );      // add conf.d directory

Direct kernel pattern (no fixture)

For tests not using the TestNG fixture machinery:

try ( Kernel kernel = new Kernel( Module.CONFIGURATION.urlsFromClassPath() ) ) {
    kernel.start( Map.of( "boot.main", "my-module" ) );
    MyService svc = kernel.serviceOfClass2( MyService.class );
    // assertions
}

Production Boot

Boot.main is the production entry point. It creates a Kernel from all classpath module descriptors and starts it.

java -cp <classpath> oap.application.Boot start \
  --config /etc/myapp/application.conf \
  --config-directory /etc/myapp/conf.d     # optional; defaults to <config parent>/conf.d

SIGINT and SIGTERM both trigger a graceful kernel.stop() followed by System.exit(0).

Boot.terminated is a public volatile boolean that becomes true when shutdown begins. Useful for polling in application-level shutdown hooks.

Error Reference

Message Cause
boot.main must contain at least one module name boot.main is empty or missing in application.conf
<url>: module.name is blank A module file has no name field
unknown application configuration module: X application.conf references a module not found on the classpath
unknown application configuration services: M.[S] application.conf overrides a service that does not exist in module M
main.boot: unknown module name 'X' boot.main names a module not found on the classpath
module name X does not match ... Module name contains characters outside [A-Za-z\-_0-9]
service name X does not match ... Service name contains illegal characters
failed to initialize service: M:S. implementation == null Service block has no implementation field
[M:*] dependencies are not enabled Module's dependsOn target is disabled
[M:S] dependencies are not enabled. Required service [X] is disabled A service parameter references a disabled service
cyclic dependency detected Module-level dependency cycle
services cyclic dependency detected Service-level dependency cycle
No implementation has been declared for the abstract service <M.S> abstract=true with no concrete impl registered and no default
No implementation specified for abstract service <M.S> ... Available implementations [...] Concrete impls exist but none was selected in application.conf
Service <M.S> has an abstract implementation, but the "abstract = true" property is missing Interface/abstract class used without abstract = true
Unknown service X in reference <modules.M.X> Abstract service default or application.conf assignment references a non-existent service
M:S Service X is already registered Two enabled services in the same module have the same name
for S listening object <ref> is not found listen reference does not resolve to a known service
listener L should have method addLListener in <ref> listen.L target has no addLListener(T) method

oap-stdlib

Core utility library for the OAP platform. Provides serialization, reflection, file I/O, collections, concurrency primitives, and identifier abstractions used across all OAP modules.

Packages

Package Contents
oap.json Binder — JSON/HOCON/YAML/XML/BSON serializer
oap.reflect Reflect, Reflection, TypeRef, Coercions
oap.id Identifier, StringIdentifier, IntIdentifier
oap.io Files, IoStreams, Resources, ContentReader, ContentWriter
oap.util Stream, Cuid, Dates, Result, Lists, Maps, Sets, Strings, Pair
oap.concurrent Executors, Threads, Scheduler, Stopwatch, LimitedTimeExecutor
oap.net Inet, IpRangeTree
oap.dictionary Dictionary

oap.json.Binder

Pre-configured Jackson ObjectMapper wrappers. All instances are thread-safe singletons.

Static instances

Instance Format Notes
Binder.json JSON Standard serializer; skips nulls
Binder.jsonWithTyping JSON Embeds @class type info for polymorphic deserialization
Binder.hocon HOCON Resolves ${?ENV_VAR} and system properties
Binder.hoconWithoutSystemProperties HOCON No system property substitution
Binder.yaml YAML
Binder.xml XML
Binder.xmlWithTyping XML With type info
Binder.bson BSON For MongoDB codecs

Serialization

// Object → String
String json = Binder.json.marshal( order );
String pretty = Binder.json.marshalWithDefaultPrettyPrinter( order );

// Object → Path (auto-detects encoding from extension)
Binder.json.marshal( Path.of( "/data/order.json.gz" ), order );

// Streaming JSON array to OutputStream
Binder.json.marshal( outputStream, List.of( order1, order2 ) );

Deserialization

Order order = Binder.json.unmarshal( Order.class, jsonString );
Order order = Binder.json.unmarshal( Order.class, path );
Order order = Binder.json.unmarshal( Order.class, url );
Order order = Binder.json.unmarshal( Order.class, inputStream );

// Generic types — use TypeRef
List<Order> orders = Binder.json.unmarshal( new TypeRef<List<Order>>() {}, jsonString );

Partial update

// Apply HOCON-format overrides to an existing object (preserves unmentioned fields)
Binder.update( order, Map.of( "status", "SHIPPED" ) );
Binder.update( order, "{ status: SHIPPED, total: 99.99 }" );

Dynamic config binders

// Parse HOCON with extra fallback properties merged in
Binder b = Binder.hoconWithConfig( Map.of( "host", "localhost", "port", 8080 ) );
MyConfig cfg = b.unmarshal( MyConfig.class, "classpath:config.conf" );

Jackson customization

Extra Jackson modules are registered by listing them in META-INF/jackson.modules (one class name per line). The default configuration:

  • Field visibility: ANY (no getters required)
  • Accepts case-insensitive property names
  • Accepts single-quoted strings
  • Skips null input values on deserialization
  • Omits null fields on serialization
  • Joda-Time, JDK8, JavaTime modules registered

oap.reflect.TypeRef<T>

Java generic type token. Use it anywhere a Class<T> cannot carry generic parameters.

TypeRef<List<Order>> ref = new TypeRef<List<Order>>() {};
List<Order> orders = Binder.json.unmarshal( ref, json );

// Also accepted by Reflect
Reflection r = Reflect.reflect( ref );

oap.reflect.Reflect / Reflection

OAP's cached reflection layer, built on Guava TypeToken.

Reflection r = Reflect.reflect( Order.class );

// Construct
Order order = r.newInstance();
Order order = r.newInstance( Map.of( "id", "o-1", "total", 42 ) );

// Fields
Reflection.Field f = r.field( "status" ).orElseThrow();
f.set( order, "SHIPPED" );
Object v = f.get( order );

// Methods
r.method( "validate" ).ifPresent( m -> m.invoke( order ) );

// Iterate all declared fields
r.fields.forEach( field -> System.out.println( field.name() + " : " + field.type() ) );

Coercions.basic() is the default type-coercion registry (String→int, String→enum, etc.) used by the Kernel when wiring service parameters.

Optional constructor parameters

A constructor parameter can be omitted from the args map passed to newInstance if it is annotated @javax.annotation.Nullable or @com.fasterxml.jackson.annotation.JsonProperty(required = false) — it is then passed as null. All other parameters must still be present as keys in args.

public Order( String id, @Nullable String note ) { ... }

Order o = r.newInstance( Map.of( "id", "o-1" ) ); // note == null, no exception

Parameter aliasing via @JsonProperty

@JsonProperty("xxx") (or @JsonProperty(value = "xxx")) renames the lookup key for a constructor parameter — newInstance reads it from args.get("xxx") instead of the Java parameter name.

Note: JsonProperty.required() defaults to false, so a plain @JsonProperty("xxx") also makes the parameter optional (see above) — pass required = true if the aliased parameter must still be present in args.

public Order( @JsonProperty( value = "order_id", required = true ) String id ) { ... }

Order o = r.newInstance( Map.of( "order_id", "o-1" ) ); // id == "o-1"

@JsonAlias({ "a", "b" }) adds further acceptable lookup keys on top of the parameter's existing name/@JsonProperty value — newInstance accepts args under any of them:

public Order( @JsonAlias( { "order_id", "orderId" } ) String id ) { ... }

r.newInstance( Map.of( "id", "o-1" ) );       // still matches (Java parameter name)
r.newInstance( Map.of( "order_id", "o-1" ) ); // also matches
r.newInstance( Map.of( "orderId", "o-1" ) );  // also matches

oap.id.Identifier<I, T>

Strategy interface that extracts, generates, and converts the ID of a data object.

// Explicit getter+setter (most common)
Identifier<String, Order> id = Identifier
    .forId( o -> o.id, ( o, newId ) -> o.id = newId )
    .suggestion( o -> o.customerName )   // derive initial id from this field
    .length( 10 )                        // max generated id length
    .build();

// Derive from a JPath expression
Identifier<String, Order> id = Identifier.<Order>forPath( "$.id" ).build();

// Use @Id annotation on the field
Identifier<String, Order> id = Identifier.<Order>forAnnotation().build();

Identifier.generate(base, length, conflict, maxAttempts, options) — slug generator with deconfliction:

  • Option.COMPACT — removes vowels from the base (shorter slugs)
  • Option.FILL — pads with X to reach length

oap.io.Files

Static file system utilities.

// Read
String text = Files.readString( path );
String text = Files.readString( path, encoding );
byte[] bytes = Files.read( path, Encoding.GZIP, ContentReader.ofBytes() );

// Write
Files.writeString( path, Encoding.PLAIN, "hello" );
Files.writeString( path, Encoding.GZIP, "hello", /* append */ false );

// Glob matching (Ant-style wildcards)
List<Path> found = Files.wildcard( basePath, "**/*.json" );
List<Path> found = Files.wildcard( basePath, "logs/*.log", "logs/*.log.gz" );

// Directory operations
Files.ensureFile( path );        // creates parent directories; does not create the file
Files.ensureDirectory( path );   // creates the directory and all parents
Files.delete( path );            // recursive delete
Files.copyDirectory( src, dest );
Files.move( src, dest );         // atomic rename where possible

// Metadata
long ts = Files.getLastModifiedTime( path );    // epoch ms
boolean exists = Files.exists( path );

// Hashed subdirectory (distributes many files across a 3-level tree)
Path deep = Files.deepPath( basePath, filename );

oap.io.IoStreams

Stream I/O with transparent compression support.

Encoding

Encoding.PLAIN    // no compression
Encoding.GZIP     // gzip
Encoding.BZIP2    // bzip2
Encoding.LZ4      // LZ4
Encoding.ZSTD     // Zstandard
Encoding.ZIP      // ZIP

// Auto-detect from path or URL extension
Encoding enc = Encoding.from( path );
Encoding enc = Encoding.from( url );

Reading

InputStream in = IoStreams.in( path );
InputStream in = IoStreams.in( path, Encoding.GZIP );

Stream<String> lines = IoStreams.lines( path );
Stream<String> lines = IoStreams.lines( path, Encoding.GZIP );
Stream<String> lines = IoStreams.lines( url );
Stream<String> lines = IoStreams.lines( inputStream );

Writing

OutputStream out = IoStreams.out( path );
OutputStream out = IoStreams.out( path, Encoding.GZIP );
OutputStream out = IoStreams.out( path, Encoding.GZIP, /* append */ true );

IoStreams.write( path, Encoding.GZIP, "text content" );
IoStreams.write( path, Encoding.GZIP, inputStream );
IoStreams.write( path, Encoding.PLAIN, lineStream );  // Stream<String>, one line per element

oap.io.Resources

Classpath resource loading relative to a context class.

// Single resource
Optional<URL>  url  = Resources.url( MyClass.class, "config.conf" );
Optional<Path> path = Resources.filePath( MyClass.class, "config.conf" );

// All resources with this name across all jars (useful for META-INF aggregation)
List<URL>  urls  = Resources.urls( "META-INF/services/MyService" );
List<Path> paths = Resources.filePaths( MyClass.class, "schemas" );

// Read content
Optional<String> text = Resources.read( MyClass.class, "query.sql", ContentReader.ofString() );
Stream<String>   lines = Resources.lines( "META-INF/jackson.modules" );

oap.io.content.ContentReader / ContentWriter

Typed I/O adapters passed to Files.read(), IoStreams.write(), and cloud storage APIs.

ContentReader factories

Factory Returns
ContentReader.ofString() String (UTF-8)
ContentReader.ofBytes() byte[]
ContentReader.ofLines() List<String>
ContentReader.ofLinesStream() Stream<String>
ContentReader.ofInputStream() InputStream (caller must close)

Chain readers with .andThen(fn):

ContentReader<MyObj> r = ContentReader.ofString()
    .andThen( s -> Binder.json.unmarshal( MyObj.class, s ) );

ContentWriter factories

Factory Writes
ContentWriter.ofString() String → UTF-8 bytes
ContentWriter.ofBytes() byte[] pass-through
ContentWriter.ofJson() Any object → JSON bytes via Binder.json
ContentWriter.ofObject() Java serialization (ObjectOutputStream)

oap.util.Cuid

Cluster-unique identifier — time-based, monotonic, embeds the local IP address.

// Production: globally unique, embeds timestamp + local IP
String id   = Cuid.UNIQUE.next();       // e.g. "0000018F3A2B1C00C0A80101"
long   idL  = Cuid.UNIQUE.nextLong();
String last = Cuid.UNIQUE.last();       // last generated (no increment)

// Parse a Cuid back to components
Cuid.UniqueCuid.Info info = Cuid.UniqueCuid.parse( id );
// info.time  → DateTime (UTC)
// info.ip    → int[4]
// info.count → per-millisecond counter

// Tests: deterministic counter starting at seed
Cuid counter = Cuid.incremental( 1 );
counter.next();     // "1"
counter.next();     // "2"

oap.util.Dates

Joda-Time utilities. All operations use UTC unless otherwise noted.

Formatters

Constant Pattern Example
Dates.FORMAT_MILLIS yyyy-MM-dd'T'HH:mm:ss.SSS 2024-06-01T14:30:00.000
Dates.FORMAT_SIMPLE yyyy-MM-dd'T'HH:mm:ss 2024-06-01T14:30:00
Dates.FORMAT_DATE yyyy-MM-dd 2024-06-01 
String s = Dates.formatDateWithMillis( DateTime.now() );
String s = Dates.FORMAT_DATE.print( dt );

Result<DateTime, Exception> r = Dates.parseDateWithMillis( "2024-06-01T14:30:00.000" );
Result<DateTime, Exception> r = Dates.parseDate( "2024-06-01T14:30:00" );
DateTime now   = Dates.nowUtc();
DateTime today = Dates.nowUtcDate();   // time zeroed to 00:00:00.000

Duration constants (return milliseconds as long)

Dates.s( 30 )   // 30 seconds in ms
Dates.m( 5 )    // 5 minutes in ms
Dates.h( 2 )    // 2 hours in ms
Dates.d( 7 )    // 7 days in ms
Dates.w( 2 )    // 2 weeks in ms

String human = Dates.durationToString( Dates.h(1) + Dates.m(30) ); // "1h 30m"

Controllable clock (for tests)

Dates.setTimeFixed( 2024, 6, 1, 14, 30, 0 );   // freeze at 14:30:00 UTC
Dates.incFixed( Dates.h( 1 ) );                  // advance by 1 hour
DateTimeUtils.setCurrentMillisSystem();           // restore real clock

oap.util.Stream<E>

OAP's extended stream — wraps java.util.stream.Stream and adds extra operations.

// Factory methods
Stream<T> s = Stream.of( collection );
Stream<T> s = Stream.of( iterator );
Stream<T> s = Stream.of( enumeration );
Stream<T> s = Stream.traverse( initialState, nextFn );  // iterator-style generator

// Extra intermediates
stream.takeWhile( predicate )          // stop at first non-matching element
stream.grouped( batchSize )           // → Stream<List<E>> in fixed-size batches
stream.grouped( classifier )         // → BiStream<K, List<E>> grouped by key
stream.zip( otherStream, zipper )    // pair-wise transform into a new type
stream.zip( otherStream )            // → BiStream<E, B>

// Extra terminals
List<E>  list = stream.toList();
Set<E>   set  = stream.toSet();
Map<K,V> map  = stream.toMap( keyFn, valueFn );

oap.util.Lists / Maps / Sets / Strings

Static utility classes.

// Lists
List<B>      mapped   = Lists.map( list, fn );
List<T>      filtered = Lists.filter( list, pred );
List<T>      concat   = Lists.concat( listA, listB );
List<T>      reversed = Lists.reverse( list );
Optional<T>  head     = Lists.head( list );

// Maps
Map<K,V>           filtered = Maps.filter( map, ( k, v ) -> pred );
List<R>            asList   = Maps.toList( map, ( k, v ) -> ... );
LinkedHashMap<K,V> linked   = Maps.toLinkedHashMap( list, keyFn, valueFn );

// Sets
Set<T> intersection = Sets.intersection( setA, setB );
Set<T> union        = Sets.union( setA, setB );
Set<T> difference   = Sets.difference( setA, setB );

// Strings
String result = Strings.substitute( "Hello ${name}!", Map.of( "name", "World" ) );
String sorted = Strings.sortLines( multilineString );
byte[] bytes  = Strings.toByteArray( str );
String hex    = Strings.toHexString( bytes );

oap.util.Result<S, F>

Typed success/failure without exceptions.

Result<Order, String> r = Result.success( order );
Result<Order, String> r = Result.failure( "not found" );

// Wrap a throwing supplier — catches all Throwable
Result<Order, Throwable> r = Result.catching( () -> orderService.find( id ) );

// Query
boolean ok     = r.isSuccess();
Order   order  = r.successValue;
String  reason = r.failureValue;

// Transform
Result<String, String>    r2 = r.mapSuccess( o -> o.id );
Result<Order, Throwable>  r3 = r.mapFailure( msg -> new RuntimeException( msg ) );

// Branch
r.ifSuccess( o -> log.info( "ok: {}", o.id ) )
 .ifFailure( e -> log.warn( "failed: {}", e ) );

// Terminate
Optional<Order> opt   = r.toOptional();
Order           order = r.orElse( defaultOrder );
Order           order = r.orElseThrow( msg -> new RuntimeException( msg ) );

oap.net.Inet

Optional<InetAddress> ip   = Inet.getLocalIp();
String                host = Inet.hostName();

oap.concurrent.Executors

// Named scheduled thread pool
ScheduledExecutorService exec = Executors.newScheduledThreadPool( 4, "my-service" );

// Named single-thread executor
ExecutorService exec = Executors.newSingleThreadExecutor( "my-worker" );

Thread names include the pool name for easy identification in thread dumps and profilers.

oap-http

HTTP server and client infrastructure for the OAP platform. Provides an Undertow-based server with named ports, a high-performance non-blocking pipeline, Prometheus metrics exporters, and test utilities.

Sub-modules

Module Description Depends on
oap-http NioHttpServer, OapHttpClient, HealthHttpHandler, HttpServerExchange —
oap-pnio-v3 High-performance non-blocking pipeline (PnioHttpHandler, PnioExchange) oap-http
oap-http-prometheus Prometheus scrape endpoint, JVM metrics, application info exporter oap-http
oap-http-test HttpAsserts, HttpServerExchangeStub, MockHttpContext oap-http

Quick start

1. Add oap-http to your module's dependsOn:

name = my-module
dependsOn = [oap-http]

2. Reference the server and bind a handler:

services {
  my-handler {
    implementation = com.example.MyHandler
    parameters {
      server = <modules.oap-http.oap-http-server>
    }
  }
}
public class MyHandler implements HttpHandler {
    public MyHandler( NioHttpServer server ) {
        server.bind( "/api/hello", this );
    }

    @Override
    public void handleRequest( HttpServerExchange exchange ) {
        exchange.responseBody( "hello" );
    }
}

The endpoint is available at GET http://localhost:8080/api/hello.

See oap-http for the full server reference.

Optional add-ons

Need Add module
High-performance non-blocking pipeline oap-pnio-v3
Prometheus metrics scrape endpoint oap-http-prometheus
HTTP assertions in tests oap-http-test

oap-ws

HTTP web service framework for the OAP platform. Provides annotation-driven endpoint declaration, session management, interceptors, validation, OpenAPI generation, SSO/JWT security, and file upload/download — all wired through the OAP Kernel with zero servlet-container boilerplate.

Sub-modules

Module Description Depends on
oap-ws Core framework: @WsMethod, @WsParam, WebServices, SessionManager, validation oap-http
oap-ws-admin-ws Built-in admin endpoints: log level control, JPath queries, JSON schema lookup oap-ws
oap-ws-api-api Shared API descriptor contracts (Info, @OpenapiIgnore) oap-ws
oap-ws-api-ws HTTP endpoint that exposes the service registry as JSON (GET /system/api) oap-ws
oap-ws-file-ws File upload and download over HTTP with multi-bucket storage oap-ws
oap-ws-openapi Core OpenAPI 3.x generation library (OpenapiGenerator, WebServicesWalker) oap-ws
oap-ws-openapi-ws HTTP endpoint that serves the generated OpenAPI spec (GET /system/openapi) oap-ws, oap-ws-api-ws
oap-ws-openapi-maven-plugin Maven plugin to generate swagger.json / YAML at build time —
oap-ws-sso-api SSO contracts + interceptors: @WsSecurity, JWT, API key, throttle-login —
oap-ws-sso AbstractSecureWS base class for secured web services oap-ws-sso-api
oap-ws-test TestNG assertion helpers for validation errors oap-ws

Quick start

1. Add oap-ws to your module's dependsOn:

name = my-module
dependsOn = [oap-ws]

2. Annotate your service class and register it with a ws-service block:

public class HelloWS {
    @WsMethod( path = "/hello", method = HttpMethod.GET )
    public String hello( @WsParam( from = From.QUERY ) String name ) {
        return "Hello, " + name + "!";
    }
}
services {
  hello-ws {
    implementation = com.example.HelloWS
    ws-service.path = api
  }
}

3. The endpoint is available at GET /api/hello?name=World.

See the oap-ws module for the full reference.

Optional add-ons

Need Add module
Runtime API introspection oap-ws-api-ws
OpenAPI / Swagger spec served at runtime oap-ws-openapi-ws
OpenAPI spec generated at build time oap-ws-openapi-maven-plugin
JWT / API-key authentication (query params or headers) oap-ws-sso-api
File upload / download oap-ws-file-ws
Admin (log levels, JPath) oap-ws-admin-ws

oap-jpath

JPath expression language for navigating Java objects and maps using reflection. Expressions are parsed by an ANTLR4 grammar and evaluated against a variable map, traversing public and private fields, calling methods, and indexing arrays or lists — all in a single ${…} expression.

Expression syntax

Every JPath expression is wrapped in ${…}. The first segment names a variable from the provided map; subsequent segments are chained with ..

${variable}
${variable.field}
${variable.field.nestedField}
${variable.method()}
${variable.method("arg", 2)}
${variable.array[0]}
${variable.list[1].field}

Path segment types

Form Example Resolves via
identifier name Field access — public or private, via reflection
name(args…) getLabel("x", 2) Method call — public or private, via reflection
name[n] items[1] Array element or List.get(n)

Segments can be chained freely:

${order.lines[0].product.getPrice("USD")}

Method arguments

Methods accept string literals and decimal integer literals as arguments.

Literal Example Parsed as
String "hello" String
Decimal integer 42 Parsed as Long, auto-coerced to the target parameter type (int, long, float, double, short, byte)

API

JPath.evaluate

StringBuilderJPathOutput output = new StringBuilderJPathOutput();

JPath.evaluate(
    "${user.address.city}",
    Map.of( "user", user ),
    output
);

String result = output.toString();

Static shorthand — builds a JPath instance and evaluates in one call. For repeated evaluation against the same variable set, construct a JPath instance directly:

JPath jpath = new JPath( Map.of( "user", user ) );

jpath.evaluate( "${user.name}", output );
output.reset();
jpath.evaluate( "${user.email}", output );

StringBuilderJPathOutput

Built-in JPathOutput implementation that collects results into a StringBuilder.

Method Description
toString() Returns the accumulated string value
reset() Clears the buffer for re-use

Examples

// Simple variable lookup
JPath.evaluate( "${id}", Map.of( "id", 42 ), output );
// → "42"

// Nested field access (public field)
JPath.evaluate( "${order.status}", Map.of( "order", order ), output );

// Private field access
JPath.evaluate( "${bean.internalState}", Map.of( "bean", bean ), output );

// Private method call
JPath.evaluate( "${bean.computeScore()}", Map.of( "bean", bean ), output );

// Method with string argument
JPath.evaluate( "${bean.format(\"prefix\")}", Map.of( "bean", bean ), output );

// Method with multiple arguments (string + integer)
JPath.evaluate( "${bean.pad(\"x\", 5)}", Map.of( "bean", bean ), output );

// Array element access
JPath.evaluate( "${data.scores[2]}", Map.of( "data", data ), output );

// List element + field chain
JPath.evaluate( "${order.lines[0].productName}", Map.of( "order", order ), output );

// Chaining Java API calls
JPath.evaluate( "${map.keySet().stream().count()}", Map.of( "map", map ), output );

Custom output

JPathOutput is a @FunctionalInterface. Implement it to collect typed values without converting to a string:

List<Object> collected = new ArrayList<>();

JPathOutput collector = pointer -> collected.add( pointer.get() );

JPath.evaluate( "${item.price}", Map.of( "item", item ), collector );

BigDecimal price = (BigDecimal) collected.get( 0 );

The Pointer passed to write is one of:

Implementation get() returns
ObjectPointer<T> The resolved object
MapPointer The resolved Map
NullPointer null

Errors

Exception Thrown when
PathNotFoundException A field or method named in the expression does not exist on the target object
ReflectionException Reflection access fails (e.g., module access denied)

See also

  • JPathWS — exposes JPath evaluation over the live Kernel service tree as a JSON HTTP endpoint.
  • InspectorWS — browsable HTML UI built on top of the same JPath queries.

oap-formats

Format processing modules for the OAP platform: template engine, TSV/CSV, JSON schema validation, and log streaming.

Sub-modules

Module Description
oap-template Compile-time template engine — parses once, compiles to Java, renders at near-native speed
oap-template-test TemplateEngineFixture — TestNG fixture for template engine tests
oap-json JSON schema validation (HOCON format) and structural diff
oap-tsv TSV/CSV parsing, streaming, and printing
oap-logstream High-throughput transactional log streaming to time-bucketed gzip files

oap-statsdb

Distributed, in-memory statistics database for the OAP platform. Data is organized as a typed key hierarchy — each level of the tree holds a Node.Value that knows how to merge itself with another value of the same type. Parent nodes optionally aggregate over their children after each update.

Architecture

StatsDBNode (process A)          StatsDBNode (process B)
  update("k1","k2", v -> v.n++)    update("k1","k3", v -> v.n++)
  sync() ─────────────────────┐    sync() ────────────────────┐
                               ▼                              ▼
                        StatsDBMaster (in-memory tree)
                          k1 → MockChild (aggregate)
                            k2 → MockValue
                            k3 → MockValue
                          ▼ (periodically)
                        StatsDBStorage (MongoDB / NULL)

In a single-process deployment, use StatsDBMaster directly without a StatsDBNode.

Sub-modules

Module Description Depends on
oap-statsdb-common Core: Node.Value, Node.Container, NodeSchema, StatsDB API —
oap-statsdb-master StatsDBMaster, StatsDBStorage, MongoDB persistence, message listener oap-statsdb-common
oap-statsdb-node StatsDBNode, StatsDBTransport, message-based sync transport oap-statsdb-common
oap-statsdb-test StatsDBTransportMock for integration tests oap-statsdb-master, oap-statsdb-node

Data model

Node.Value<T>

The value stored at each tree node. Must implement merge(T other) — called when a sync from a remote node arrives — and Serializable.

public class Counters implements Node.Value<Counters> {
    public long requests;
    public long errors;

    @Override
    public Counters merge( Counters other ) {
        requests += other.requests;
        errors   += other.errors;
        return this;
    }
}

Node.Container<T, TChild>

A value at an intermediate tree level that rolls up metrics from its children. aggregate(List<TChild>) is called automatically after every update on any descendant.

public class RollupCounters implements Node.Container<RollupCounters, Counters> {
    public long totalRequests;

    @Override
    public RollupCounters merge( RollupCounters other ) {
        // merge is additive — called when syncing from remote nodes
        return this;
    }

    @Override
    public RollupCounters aggregate( List<Counters> children ) {
        totalRequests = children.stream().mapToLong( c -> c.requests ).sum();
        return this;
    }
}

Mark computed fields @JsonIgnore if they should not be persisted (they are re-derived from children on load).

NodeSchema

Declares the Node.Value class at each key level, ordered from root to leaf.

NodeSchema schema = new NodeSchema(
    nc( "endpoint",  RollupCounters.class ),  // level 0 — root
    nc( "date",      Counters.class )          // level 1 — leaf
);

nc(String key, Class<T>) is a static factory on NodeSchema.

Register value classes in oap-module.oap so the JSON binder can deserialize them:

configurations = [
  {
    loader = oap.json.TypeIdFactory
    config {
      counters         = com.example.Counters
      rollup-counters  = com.example.RollupCounters
    }
  }
]

StatsDB API

All update and query methods are available on both StatsDBMaster and StatsDBNode.

Writing

// 1-key update (leaf at level 0)
db.<Counters>update( "endpoint-a", v -> v.requests++ );

// 2-key update (leaf at level 1)
db.<Counters>update( "endpoint-a", "2024-06-01", v -> {
    v.requests++;
    v.errors++;
} );

// Up to 5 keys supported
db.<Counters>update( k1, k2, k3, k4, k5, v -> v.requests++ );

Reading

// Get value at a path (returns null if not present)
Counters c = db.get( "endpoint-a", "2024-06-01" );

// Get all child values under a prefix
Stream<Counters> daily = db.children( "endpoint-a" );

Typed select streams

Use select2() … select5() to stream over the full tree with typed key-value tuples:

// 2-level tree: (id1, v1) → (id2, v2)
db.<RollupCounters, Counters>select2().forEach( row -> {
    System.out.println( row.id1 + " " + row.id2 + " requests=" + row.v2.requests );
} );

// 3-level tree
db.<T1, T2, T3>select3().forEach( row -> { … } );
// also select4(), select5()
Method Fields
select2() id1, v1, id2, v2
select3() id1, v1, id2, v2, id3, v3
select4() id1, v1, id2, v2, id3, v3, id4, v4
select5() id1, v1, id2, v2, id3, v3, id4, v4, id5, v5

Clearing

db.removeAll();  // clears in-memory state only

Quick start — single process

NodeSchema schema = new NodeSchema(
    nc( "endpoint", RollupCounters.class ),
    nc( "date",     Counters.class )
);

try( StatsDBMaster master = new StatsDBMaster( schema, StatsDBStorage.NULL ) ) {
    master.<Counters>update( "search", "2024-06-01", v -> v.requests += 5 );
    master.<Counters>update( "search", "2024-06-02", v -> v.requests += 3 );

    // Roll-up is automatic
    assertThat( master.<RollupCounters>get( "search" ).totalRequests ).isEqualTo( 8 );
}

oap-message

Reliable, durable HTTP message delivery for the OAP platform. The sender buffers messages to disk when the network is unavailable and retries until acknowledged. The server deduplicates by MD5 so retries are always safe to replay.

Architecture

MessageSender (client process)
  send(type, data)
    │
    ├─ in-memory queue ──► syncMemory() ──► POST /messages ──► MessageHttpHandler
    │                                                                │
    └─ disk (on shutdown/failure)                                   ├─ MD5 dedup (MessageHashStorage)
         syncDisk() reloads on restart                              │
                                                                    └─ MessageListener.run(...)
                                                                         → short status

Wire protocol

Request (client → server)

Field Type Description
message type byte User-defined type identifier (0–200)
version short Message schema version
client ID long Unique sender ID (per MessageSender instance)
MD5 byte[16] MD5 digest of the payload
reserved byte[8] Reserved, always zero
data size int Payload length in bytes
payload byte[N] Message body

Response (server → client)

Field Type Description
protocol version byte Always 1
client ID long Echoed from request
MD5 byte[16] Echoed from request
reserved byte[8] Reserved
status short See status codes below

Status codes

Constant Value Meaning
STATUS_OK 0 Processed successfully
STATUS_UNKNOWN_ERROR 1 Processing failed — client will retry
STATUS_UNKNOWN_ERROR_NO_RETRY 2 Processing failed — client drops the message
STATUS_UNKNOWN_MESSAGE_TYPE 100 No listener registered for this type — client drops the message
STATUS_ALREADY_WRITTEN 101 Duplicate — server already processed this MD5; treated as success by the client

Custom status codes (causing retry) can be registered in META-INF/oap-messages.properties using the map.* prefix — see oap-message-server.

Sub-modules

Module Description Depends on
oap-message-client MessageSender — durable send queue with disk persistence oap-http
oap-message-server MessageHttpHandler, MessageListener, MessageListenerJson oap-http
oap-message-test MessageListenerMock, MessageListenerJsonMock, MessageSenderUtils oap-message-client, oap-message-server

oap-storage

Persistent, in-memory storage layer for the OAP platform. Objects are kept in a ConcurrentHashMap-backed MemoryStorage and optionally synced to MongoDB or cloud object stores.

Architecture

                  ┌─────────────────────────────┐
                  │      MemoryStorage<Id,Data>  │
                  │  (ConcurrentHashMap + Lock)  │
                  └──────────┬──────────────────┘
                             │  TransactionLog (change log)
              ┌──────────────┴──────────────────┐
              │                                 │
   MongoPersistence<I,T>          ReplicationMaster / RemoteStorage
   (periodic bulk write,          (diff-based replication
    change stream watch)           between nodes)

Cloud storage (FileSystem) is a separate, stateless API over object stores — it does not integrate with MemoryStorage.

Sub-modules

Module Description Depends on
oap-storage Storage<Id,Data>, MemoryStorage, Metadata, DataListener, Migration oap-stdlib
oap-storage-mongo MongoPersistence, MongoClient, MongoIndex, Version oap-storage
oap-storage-cloud FileSystem, CloudURI, FileSystemConfiguration, FileSystemCloudApi oap-stdlib
oap-storage-cloud-aws-s3 AWS S3 backend (s3:// scheme) oap-storage-cloud
oap-storage-cloud-ftp FTP/FTPS backend (ftp:///ftps:// schemes) oap-storage-cloud
oap-storage-cloud-smb SMB/CIFS backend (smb:// scheme) oap-storage-cloud
oap-storage-mongo-test MongoFixture — in-memory MongoDB for tests oap-storage-mongo
oap-storage-cloud-test S3MockFixture — LocalStack S3 for tests oap-storage-cloud-aws-s3

Quick start

// Define an identifier — extracts/assigns the String key from your object
Identifier<String, MyData> id = Identifier.forId( d -> d.id, ( d, newId ) -> d.id = newId )
    .suggestion( d -> d.name )
    .build();

// In-memory store, concurrent reads and writes
MemoryStorage<String, MyData> storage = new MemoryStorage<>( id, Lock.CONCURRENT );

// Store
storage.store( new MyData( "item-1", "hello" ), "system" );

// Read
Optional<MyData> found = storage.getNullable( "item-1" );

// Update in place
storage.update( "item-1", d -> { d.name = "world"; return d; } );

// Listen to changes
storage.addDataListener( new Storage.DataListener<String, MyData>() {
    @Override
    public void updated( IdObject<String, MyData> previous, IdObject<String, MyData> updated ) {
        System.out.println( "changed: " + updated.id );
    }
} );

Cloud storage (oap-storage-cloud)

Provider-agnostic cloud object storage API for the OAP platform. A single FileSystem facade dispatches to pluggable backends (AWS S3, Google Cloud Storage, Azure Blob, local filesystem) selected by URI scheme. Depends on: oap-stdlib.

CloudURI

Every path is represented as a CloudURI, addressed by configurationId, not by backend/container directly:

fs://configurationId/path/to/object
  │         │              │
  │         │              └─ object key (no leading slash)
  │         └─ named target: resolves to a backend scheme + connection (container) via config
  └─ fixed literal scheme

The configurationId is the only thing the URI carries — which backend scheme it maps to (s3, ftp, smb, ...) and which concrete connection (bucket, host[:port], host[:port]/share, ...) it uses are both resolved from FileSystemConfiguration at call time (see below).

Scheme Backend
s3 AWS S3 (requires oap-storage-cloud-aws-s3 on classpath)
gcs Google Cloud Storage
ab Azure Blob Storage
file Local filesystem
ftp FTP (requires oap-storage-cloud-ftp on classpath)
ftps FTP over TLS (requires oap-storage-cloud-ftp on classpath)
smb SMB/CIFS (requires oap-storage-cloud-smb on classpath) 
CloudURI uri = new CloudURI( "fs://my-configuration-id/data/report-2024-06-01.json" );
// uri.configurationId = "my-configuration-id"
// uri.path            = "data/report-2024-06-01.json"

// equivalent, canonical constructor
CloudURI uri2 = new CloudURI( "my-configuration-id", "data/report-2024-06-01.json" );

// Builder-style copies
CloudURI other = uri.withPath( "data/report-2024-06-02.json" );
CloudURI otherConfigurationId = uri.withConfigurationId( "other-configuration-id" );
Migrating a legacy scheme://container/path string

FileSystem.resolve(configurationId, String) accepts the old scheme://container/path shape (as used before configurationIds existed) and tags the result with the given configurationId directly — the URI's container doesn't need to be registered in config at all:

CloudURI uri = fileSystem.resolve( "my-configuration-id", "s3://my-bucket/data/report-2024-06-01.json" );

Always throws for file://... (local paths have no container to match against — use fs://file/<path> or new CloudURI("file", path) directly). fs://... input passes straight through to new CloudURI(uri), with its embedded alias replaced by configurationId.

FileSystemConfiguration

Holds per-scheme, per-configurationId, and global-default credentials and settings. Keys follow the pattern:

fs.<scheme>.<property>[.<configurationId>]
fs.default.<property>

Looking up a property for a given (scheme, configurationId) tries, in order:

  1. fs.<scheme>.<property>.<configurationId> — configurationId-specific override
  2. fs.<scheme>.<property> — scheme-wide default
  3. fs.default.<property> — global fallback, for any property, across every scheme

fs.default.* is entirely optional — there's no required key under it, and a config with no fs.default.* at all is perfectly valid.

Key charset

Every dot-separated part of an fs.* key may contain only letters, digits, and single underscores as internal separators. No hyphens, no leading/trailing/double underscore.

  • fs.a_b.a_b.d — valid
  • fs.a.b-g — invalid (hyphen)
  • fs.a__b — invalid (double underscore)
Overriding via system properties and environment variables

Any fs.* key can also be supplied as a JVM system property or an OS environment variable, without touching the map/HOCON config — useful for ops-level overrides. Priority, highest first:

  1. Environment variable
  2. JVM system property
  3. The Map/HOCON passed to the constructor

System properties are matched by literal fs. prefix and used as-is, no translation:

java -Dfs.s3.container=override-bucket -jar app.jar

Environment variables use conventional FS_... naming and are decoded back into a dotted key: . in the key becomes a single _ in the env name, and a literal _ already in the key becomes __ in the env name.

Property Env variable
fs.a.b.d FS_A_B_D
fs.a.b.d_f FS_A_B_D__F 
export FS_S3_CONTAINER=override-bucket
ConfigurationIds

A configurationId is a named target (a backend scheme + connection). It's detected from configuration — no separate declaration list, and every caller states the configurationId it means explicitly (FileSystem has no notion of "the default one"):

  • Any configurationId is registered the moment it appears in a fs.<scheme>.container.<configurationId> key — container is the anchor property every configurationId needs to actually connect to something, so declaring it is what makes the configurationId exist.
  • A bare configurationId equal to an installed backend's scheme name (fs://ftp/..., fs://file/...) resolves implicitly with zero configurationId-related config, so single-target setups need nothing beyond the scheme-wide properties.
  • FileSystemConfiguration.required(String configurationId) validates up front that a configurationId is registered to some scheme, throwing CloudException immediately if not.
FileSystemConfiguration config = new FileSystemConfiguration( Map.of(
    // S3 credentials (apply to every configurationId on this scheme unless overridden per-configurationId)
    "fs.s3.identity",   "AKIAIOSFODNN7EXAMPLE",
    "fs.s3.credential", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "fs.s3.region",     "us-east-1",
    "fs.s3.container",  "my-bucket"
    // "s3" is both the scheme and the configurationId here (self-name convention) — no further registration needed
) );

Values support ${env.VAR_NAME} and ${system.property} substitution.

Multi-configurationId example
fs.ftp.container   = ftp.example.com:21
fs.ftp.identity    = shared-user
fs.ftp.credential  = shared-pass

# the primary account: container is repeated so this configurationId gets registered
fs.ftp.container.primary    = ftp.example.com:21

# a second account on the same server: only identity/credential differ,
# container is repeated so this configurationId gets registered
fs.ftp.container.secondary  = ftp.example.com:21
fs.ftp.identity.secondary   = other-user
fs.ftp.credential.secondary = other-pass

fs://primary/... connects as shared-user; fs://secondary/... connects to the same host as other-user.

OAP module configuration
name = my-app
dependsOn = [oap-storage-cloud]

services {
  oap-storage-cloud.oap-cloud-configuration.parameters {
    configuration {
      fs.s3.identity   = ${?AWS_ACCESS_KEY_ID}
      fs.s3.credential = ${?AWS_SECRET_ACCESS_KEY}
      fs.s3.region     = us-east-1
      fs.s3.container  = my-bucket
    }
  }
}

FileSystem

Stateless facade that routes calls to the right backend by resolving the URI's configurationId to a scheme (via FileSystemConfiguration, falling back to an installed backend's scheme name for a bare self-named configurationId). Backend instances are cached and closed with FileSystem.close(), keyed by scheme://configurationId — every backend is configurationId-scoped, since configurationId is the stable per-connection identity (this also means two configurationIds on the same S3 bucket with different credentials get independent cached clients, not a shared one).

FileSystem fs = new FileSystem( config );

// Upload
CloudURI dest = new CloudURI( "fs://my-configuration-id/reports/2024-06-01.json" );
fs.upload( dest, BlobData.builder()
    .content( jsonBytes )
    .tags( Map.of( "env", "prod" ) )
    .build() );

// Download to local path
fs.downloadFile( dest, Path.of( "/tmp/report.json" ) );

// Stream read
try( InputStream in = fs.getInputStream( dest ) ) { ... }

// Stream write
try( OutputStream out = fs.getOutputStream( dest, Map.of() ) ) { ... }

// Copy between URIs (may cross backends)
fs.copy( src, dest, Map.of( "copied", "true" ) );

// List objects
PageSet<? extends FileSystem.StorageItem> page = fs.list(
    new CloudURI( "my-configuration-id", "reports/" ),
    ListOptions.builder().maxResults( 100 ).build()
);

// Metadata only (no download)
FileSystem.StorageItem meta = fs.getMetadata( dest );
// meta.getName(), meta.getSize(), meta.getLastModified(), meta.getETag(), meta.getContentType()

// Build a CloudURI for a given configurationId
CloudURI defaultUri = fs.getDefaultURL( "my-configuration-id", "reports/today.json" );

// Migrate a legacy scheme://container/path string to a configurationId-based CloudURI
CloudURI legacyResolved = fs.resolve( "my-configuration-id", "s3://my-bucket/reports/today.json" );
Operations reference

All methods are synchronous/blocking.

Method Description
getInputStream(uri) Open object for reading
getOutputStream(uri, tags) Open object for writing
upload(uri, blobData) Write bytes / stream with optional tags
downloadFile(uri, localPath) Save object to a local file
copy(src, dest, tags) Cross-backend copy via stream
list(uri, options) List objects under a prefix; returns PageSet
getMetadata(uri) Fetch object metadata without body
blobExists(uri) Check whether an object exists
deleteBlob(uri) Delete a single object
containerExists(uri) Check whether a bucket/container exists
createContainer(uri) Create a bucket/container
deleteContainer(uri) Delete an empty bucket/container
deleteContainerIfEmpty(uri) Delete only if empty; returns boolean
getDefaultURL(configurationId, path) Build a CloudURI for the given configurationId, normalizing path separators
resolve(configurationId, legacyUri) Map a legacy scheme://container/path string onto the given configurationId
toLocalFileURI(configurationId, path) Convert a java.nio.Path or String to a fs://<configurationId>/... CloudURI for the given file configurationId

Caching

A configurationId can be declared as a read-through/write-back cache for another configurationId:

fs.<cacheScheme>.cache.get.<cacheConfigurationId> = <targetConfigurationId>[,<targetConfigurationId>...]

Declared under the cache configurationId's own scheme — same place filesystem.basedir.<id>/container.<id> live for that configurationId.

fs.file.container.fs: ""
fs.file.filesystem.basedir.fs: "/tmp"

fs.file.container.cachefs: ""
fs.file.filesystem.basedir.cachefs: "/cache"
fs.file.cache.get.cachefs: "fs"

Here cachefs (basedir /cache) is the cache for fs (basedir /tmp). The cache configurationId can be any scheme, independent of the target's scheme — a typical setup caches a remote configurationId (s3/ftp/smb) behind a local file one; the example above just uses file on both sides.

The cache-side path is namespaced by the source configurationId: fs://fs/a/b/c/file.txt caches to fs://cachefs/fs/a/b/c/file.txt (not fs://cachefs/a/b/c/file.txt), so one cache configurationId can safely back multiple distinct source configurationIds without their paths colliding.

cache.get.<cacheConfigurationId>'s value can list several target configurationIds separated by commas — fs.file.cache.get.cachefs = "fs1,fs2,fs3" makes cachefs the cache for fs1, fs2, and fs3 all at once, each still isolated under its own namespaced subpath (fs://cachefs/fs1/..., fs://cachefs/fs2/..., fs://cachefs/fs3/...).

Cache-aware operations

Only read paths that resolve a source CloudURI, plus delete, consult the cache — everything else (upload, getOutputStream, list, getMetadata, blobExists, container operations, ...) is unaffected:

Method Behavior
getInputStream(uri) Reads through the cache configurationId if one is configured for uri's configurationId
downloadFile(uri, path) Same
copy(source, dest, tags) Same, when source is a CloudURI (not a local Path/File)
deleteBlob(uri) Deletes from uri's configurationId, then also deletes the cached copy if one exists
Read-through / refresh behavior

Every cache-aware read compares metadata (getMetadata(...).getLastModified()) on the target against the cached copy:

  • No cache configured for the configurationId → behaves exactly as without caching.
  • Cache copy missing → populated by streaming from the target into the cache configurationId, then served from there.
  • Cache copy present and the target's lastModified is not newer → served straight from the cache, without touching the target's data (only its lightweight metadata was fetched).
  • Cache copy present but the target has been modified since → the cache is refreshed (re-streamed from the target) before being served.
  • Target no longer exists but a cache copy does → the (now stale, but only source-of-truth's) cached copy is served as-is rather than erroring.
Delete propagation

deleteBlob(uri) also deletes the cached copy (if any) after deleting uri itself, so a deleted file doesn't linger in the cache.

FileSystemCloudApi

Interface implemented by each backend. Register a new implementation by placing a cloud-service.properties file on the classpath:

# cloud-service.properties
s3=com.example.MyS3CloudApi

The class must have a constructor (FileSystemConfiguration, String configurationId) — each backend resolves its own connection details (bucket, host[:port], ...) from config via fileSystemConfiguration.getOrThrow(scheme, configurationId, "container"), rather than receiving them pre-parsed.

Every method is a required synchronous, blocking method.

AWS S3

Add the oap-storage-cloud-aws-s3 artifact to your dependencies. The s3:// scheme is registered automatically via cloud-service.properties — no additional wiring is needed.

Required configuration keys for S3 (each supports the configurationId-override / scheme-wide / fs.default.* fallback chain):

Key Description
fs.s3.container Bucket name
fs.s3.identity AWS access key ID
fs.s3.credential AWS secret access key
fs.s3.region AWS region (e.g. us-east-1)
fs.s3.endpoint Override endpoint URL (e.g. for LocalStack); when set, path-style access is forced automatically
fs.s3.filesystem.basedir Optional key prefix within the bucket; every object key is resolved as <basedir>/<path> and list() results are returned relative to it, same as file's filesystem.basedir

Backed by the AWS SDK v2 S3Client, built with Apache5HttpClient.

Credentials and region
  • fs.s3.identity/fs.s3.credential are optional — when either is missing, the client falls back to the AWS SDK's default credential provider chain (environment variables, ~/.aws/credentials, EC2/ECS instance profile, ...) instead of static credentials.
  • Region resolution order: fs.s3.region → the AWS_REGION environment variable → Region.AWS_GLOBAL if neither is set.
toUri()

Renders a real, fetchable HTTP(S) URL rather than an s3://bucket/key-style pseudo-URI:

  • fs.s3.endpoint set (e.g. LocalStack/MinIO) → <endpoint>/<bucket>/<path> (path-style, matching the forcePathStyle(true) the client itself uses when an endpoint override is configured).
  • No endpoint → https://s3.<region>.amazonaws.com/<bucket>/<path>, or https://s3.amazonaws.com/<bucket>/<path> when no region resolved either.
Multipart upload

getOutputStream(uri, tags) and streamed upload(uri, blobData) (an InputStream BlobData) both write through an internal MultipartUploadOutputStream, buffering up to 5 MB (PART_SIZE) per part:

  • Content that never exceeds one 5 MB buffer is sent as a single PutObject on close() — no multipart upload is started for small objects.
  • Once a second part is needed, a multipart upload is created lazily on the first buffer flush, and close() completes it with all uploaded parts.
  • Any SDK failure mid-upload aborts the in-progress multipart upload (AbortMultipartUploadRequest) before the CloudException propagates.
Other S3-specific notes
  • Unlike FTP/SMB, S3 supports object tagging — BlobData.tags (or getOutputStream's tags parameter) are attached to the object via PutObject/CreateMultipartUpload tagging, for every upload path.
  • deleteContainer(uri) lists and deletes every object in the bucket (not scoped to filesystem.basedir) before deleting the bucket itself — it ignores uri's path entirely.
  • deleteContainerIfEmpty(uri) distinguishes "bucket not empty" from other failures by matching the AWS error message text, returning false only for that case; anything else throws CloudException.

FTP

Add the oap-storage-cloud-ftp artifact to your dependencies. The ftp:// and ftps:// schemes are registered automatically via cloud-service.properties.

Unlike file, FTP/FTPS require a container: fs.ftp.container[.<configurationId>] (host[:port]) identifies the FTP server a configurationId connects to. getOrThrow throws CloudException if no container can be resolved for the configurationId.

Each distinct configurationId gets its own pooled connection set — two configurationIds pointing at different hosts (or even the same host with different credentials) never share a connection pool.

FTP control connections (TCP connect + login) are pooled per backend instance using Apache Commons Pool 2 — operations borrow a connection from the pool and return it when done instead of reconnecting/logging in on every call. Pooled connections are validated with an FTP NOOP before reuse, so idle connections dropped by the server/firewall are transparently replaced.

Required/optional configuration keys (each supports the configurationId-override / scheme-wide / fs.default.* fallback chain):

Key Description
fs.ftp.container host[:port] of the FTP server (default port 21)
fs.ftp.identity FTP username (default anonymous)
fs.ftp.credential FTP password
fs.ftp.passive_mode true/false (default true)
fs.ftp.remove_empty_folders true to delete now-empty parent directories after a blob delete (default false)
fs.ftp.pool_max_size Max pooled FTP connections per backend instance (default 8)
fs.ftp.pool_max_wait_millis Max time to wait for a pooled connection before failing, in milliseconds (default 30000)
fs.ftp.connect_timeout_millis TCP connect timeout, in milliseconds (default 30000)
fs.ftp.default_timeout_millis Timeout applied to the socket immediately after connecting, before login, in milliseconds (default 30000)
fs.ftp.so_timeout_millis Timeout while waiting for control-connection responses, in milliseconds (default 30000)
fs.ftps.tls_mode explicit (default) or implicit
fs.ftps.trust_all true to skip server certificate validation (e.g. self-signed certs in tests)
fs.ftp.filesystem.basedir Optional remote path prefix; every path is resolved as <basedir>/<path> and list() results are returned relative to it, same as file's filesystem.basedir 
CloudURI dest = new CloudURI( "fs://my-ftp-configuration-id/reports/2024-06-01.json" );
fs.upload( dest, BlobData.builder().content( jsonBytes ).build() );
Per-configurationId FTP configuration overrides

Declaring fs.ftp.container.<configurationId> registers <configurationId> and gives it its own connection target; pairing it with fs.ftp.identity.<configurationId>/fs.ftp.credential.<configurationId> gives that configurationId its own credentials too (see the multi-configurationId example above). Since the lookup mechanism probes exact key strings rather than positionally splitting stored keys, configurationId names need no dot-escaping, unlike the old per-container scheme:

fs.ftp.container.reporting-server = ftp.server1.example.com:21
fs.ftp.identity.reporting-server  = as

A configurationId-specific entry overrides fs.ftp.<property> only for that exact configurationId; other configurationIds on the same scheme keep falling back to the scheme-wide default, and ultimately to fs.default.<property>.

createContainer/deleteContainerIfEmpty always return false, and deleteContainer throws CloudException — there's no container to create or delete. FTP also has no object-tagging concept, so tags passed to upload/getOutputStream are ignored.

Streaming, auto-mkdir, and pool health
  • getInputStream/getOutputStream wrap the raw FTP data-connection stream; on close() they call completePendingCommand() and return the borrowed connection to the pool healthy, or invalidate it, based on that result — an incomplete/broken transfer drops the connection instead of poisoning future borrows.
  • getOutputStream, upload, and copy's destination side all create any missing remote parent directories automatically — no separate mkdir call needed.
  • copy(source, destination) borrows two pooled connections at once (source and destination) and streams between them — fs.ftp.pool_max_size must be at least 2 for copy to succeed without waiting on itself.
list() is not server-paginated

Unlike S3, FTP has no native continuation-token listing: list() walks the entire directory tree recursively into memory first, sorts it, and only then applies continuationToken/maxKeys as an in-memory skip/limit. Fine for moderate directory sizes; a very large tree is read in full on every list() call regardless of maxKeys.

remove_empty_folders cleanup depth

On delete, if enabled, empty parent directories are removed one level at a time, walking upward only as far as the configured filesystem.basedir (or the FTP root if no basedir is set) — basedir itself is never removed.

Lazy modification time

getMetadata(uri) and each list() result compute getLastModified() lazily and only once, via a separate FTP MDTM command (borrowing another pooled connection) issued the first time it's actually read — not eagerly during the directory walk.

SMB

Add the oap-storage-cloud-smb artifact to your dependencies. The smb:// scheme (backed by jcifs-ng) is registered automatically via cloud-service.properties.

Like FTP, SMB requires a container, but fs.smb.container[.<configurationId>] is host[:port]/share (default port 445) — the share is part of the container, not the path. fs.smb.container = fileserver:445/reports addresses share reports on fileserver:445; the object path is relative to that share. getOrThrow throws CloudException if no container (or no share segment within it) can be resolved for the configurationId.

Each configurationId gets its own backend instance holding one CIFSContext — jcifs-ng manages the underlying SMB session/connection reuse internally, so (unlike FTP) there's no separate connection-pool configuration. Two configurationIds never share a session, even if they point at the same share.

Required/optional configuration keys (each supports the configurationId-override / scheme-wide / fs.default.* fallback chain):

Key Description
fs.smb.container host[:port]/share
fs.smb.identity SMB username (default guest)
fs.smb.credential SMB password
fs.smb.domain NTLM domain/workgroup (default empty)
fs.smb.filesystem.basedir Optional path prefix within the share; every path is resolved as <basedir>/<path> and list() results are returned relative to it, same as file's filesystem.basedir 
CloudURI dest = new CloudURI( "fs://my-smb-configuration-id/reports/2024-06-01.json" );
fs.upload( dest, BlobData.builder().content( jsonBytes ).build() );

createContainer/deleteContainerIfEmpty always return false, and deleteContainer throws CloudException — SMB shares aren't created/deleted through this client. SMB has no object-tagging concept, so tags passed to upload/getOutputStream are ignored.

oap-highload

CPU affinity utility for the OAP platform. Pins the calling thread to a specific CPU core via net.openhft.affinity, reducing cross-core cache misses in high-throughput loops (network I/O, encoding, scheduling).

Affinity

A plain utility class — not a managed OAP service. Instantiate it directly wherever you need to control thread placement.

CPU set syntax

The constructor accepts a string that describes which CPU cores to use:

Expression Meaning Example → CPUs
* No affinity (disabled) * → []
n Single core 3 → [3]
n-m Inclusive range 1-3 → [1, 2, 3]
n+ Core n through the last available 4+ on 8-core → [4, 5, 6, 7]
Comma-separated Combine any of the above 1-3, 8 → [1, 2, 3, 8] 
Affinity affinity = new Affinity( "2-5" );   // cores 2, 3, 4, 5
Affinity affinity = new Affinity( "0+" );    // all cores from 0 upward
Affinity affinity = new Affinity( "*" );     // disabled — no pinning
Affinity affinity = Affinity.any();          // same as "*"

API

Method Description
set() Pin the calling thread to the next core in the set (round-robin); no-op when disabled
isEnabled() false when constructed with *; true otherwise
size() Number of CPU cores in the configured set
getCpus() Raw int[] of configured core indices

Usage

Call set() once per thread at startup, or at the top of a processing loop when you want round-robin distribution across the configured cores:

Affinity affinity = new Affinity( "4+" );   // dedicate upper cores to this pool

ExecutorService pool = Executors.newFixedThreadPool( affinity.size(), r -> {
    Thread t = new Thread( () -> {
        affinity.set();   // pin this thread before doing any work
        r.run();
    } );
    return t;
} );

When isEnabled() is false (e.g. * in config), set() is a no-op and the JVM scheduler assigns cores freely — no code path changes needed.

oap-mail

Email delivery for the OAP platform. Provides a persistent delivery queue, Velocity-based message templates, and swappable transports (SMTP, SendGrid).

Architecture

Template → Message → Mailman → MailQueue → Transport
                                    ↕
                            MailQueuePersistence
                         (file / memory / MongoDB)

Mailman runs as a supervised background thread that drains MailQueue. Failed messages are retried on a configurable schedule; messages that remain broken past brokenMessageTTL are dropped.

Sub-modules

Module Description
oap-mail Core: Message, Mailman, MailQueue, SmtpTransport, Template
oap-mail-sendgrid SendGrid REST API transport
oap-mail-mongo MongoDB-backed queue persistence
oap-mail-test TransportMock, MessageAssertion, MessagesAssertion, MailBox

oap-notification

Pub/sub notification delivery for the OAP platform. NotificationService sends/receives Notification messages through a pluggable NotificationTransport; the only transport currently provided is MQTT, via HivemqNotificationTransport.

Sub-modules

Module Description Depends on
oap-notification-client Core: NotificationTransport, NotificationService, Notification, NotificationPublish, NotificationPublishWithAcknowledge, Qos —
oap-notification-mqtt HivemqNotificationTransport — MQTT 5 transport (HiveMQ MQTT Client) oap-notification-client
oap-notification-test MosquittoFixture — Testcontainers Mosquitto broker for tests oap-notification-mqtt, oap-stdlib-test

Core API (oap-notification-client)

NotificationTransport transport = new HivemqNotificationTransport( "my-service-%rnd%", "mqtt.example.com", 1883 );
transport.start();

NotificationService notificationService = new NotificationService( transport );
notificationService.sendNotification( "/topic", Qos.AT_LEAST_ONCE, false, myMessage );      // myMessage: Serializable; retain: keep as the topic's last value for future subscribers

notificationService.subscribeToTopic( "/topic", publish -> {
    Serializable message = publish.message;   // NotificationPublish extends Notification, adds `topic`
} );

// Manual acknowledgement — the transport waits for acknowledge() before considering the message delivered
notificationService.subscribeToTopic( "/topic", true, publish -> {
    // ... process publish.message ...
    publish.acknowledge();
} );

transport.close();
  • Qos mirrors MQTT QoS levels: AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE.
  • Notification.message is serialized with polymorphic type info (TypeIdFactory, object:type property) — register message classes the same way as oap-statsdb value classes (see oap-statsdb's configurations/TypeIdFactory example) so they round-trip through JSON.
  • subscribeToTopic(topic(s), notificationConsumer) (no manualAcknowledgement arg) delivers plain NotificationPublish and auto-acknowledges as soon as the transport hands the message off. subscribeToTopic(topic(s), true, notificationConsumer) delivers a NotificationPublishWithAcknowledge instead (still typed as NotificationPublish in the callback) — the consumer must call .acknowledge() once done, or the transport redelivers the message (MQTT: for any QoS above AT_MOST_ONCE).
  • NotificationPublish.acknowledge() is callable on every delivered notification, no cast needed — it throws NotificationException unless the subscription was made with manualAcknowledgement = true (in which case the actual instance is a NotificationPublishWithAcknowledge, whose override does the real work). Calling it on a plain, auto-acknowledged delivery is a usage error.

HivemqNotificationTransport (oap-notification-mqtt)

An oap.notification.NotificationTransport implementation backed by HiveMQ MQTT Client (MQTT 5). Publishes/subscribes Notification messages as JSON over MQTT topics.

Wired via oap-module.oap:

name = oap-notification-mqtt
dependsOn = oap-notification

services {
  mqtt-notification-transport {
    implementation = oap.notification.mqtt.HivemqNotificationTransport
    parameters {
      identifier = "my-service-%rnd%"
      host = "mqtt.example.com"
      port = 1883
    }
    supervision.supervise = true
  }
}

Constructor parameters

Parameter Description
identifier MQTT client identifier. Supports the %rnd% placeholder (see below).
host MQTT broker host
port MQTT broker port

Tunable fields

Field Default Description
connectTimeout 10 s Timeout for start()/close()'s connect/disconnect handshake
publishTimeout 1 s Timeout for publish()/subscribe() calls

%rnd% identifier placeholder

MQTT brokers require each connected client to use a unique identifier — connecting a second client with an identifier already in use disconnects the first one. %rnd% in identifier is replaced at construction time with 5 random letters (RandomStringUtils.insecure().nextAlphabetic(5)), so a single configured identifier (e.g. "my-service-%rnd%") can be reused across multiple instances or replicas of the same service without them colliding on the broker.

new HivemqNotificationTransport( "my-service-%rnd%", host, port ).getIdentifier();
// → e.g. "my-service-qKPzr"

%rnd% is a plain String.replace substitution, computed once per constructor call — an identifier without %rnd% is passed through unchanged, and every occurrence of %rnd% (if there's more than one) is replaced with the same random value.

Testing

oap-notification-test provides:

  • MosquittoFixture — starts a real Mosquitto broker in a Testcontainers container for integration tests (see MosquittoNotificationServiceTest).
  • HivemqNotificationTransportTest — plain unit tests for the %rnd% identifier substitution (no broker required).

oap-maven-plugin

Build-time code generation and packaging utilities for OAP projects. All goals share the prefix oap.

Goals

Goal Module Phase Description
oap:generate oap-dictionary-maven generate-sources Generate Java enums from dictionary JSON/HOCON files
oap:startup-scripts oap-application-maven prepare-package Generate OS service scripts (systemd, sysvinit, shell)
oap:copy oap-maven prepare-package Copy file sets into a directory with optional property filtering

About

Open Application Platform

Resources

Readme
MIT license
Activity
Custom properties

Stars

15 stars

Watchers

9 watching

Forks

6 forks
Report repository

Releases

Packages

Used by

Contributors

Languages