diff --git a/bom/pom.xml b/bom/pom.xml
index c22ded21a..b43f9fb4e 100644
--- a/bom/pom.xml
+++ b/bom/pom.xml
@@ -503,6 +503,75 @@
helidon-common-metrics${helidon.version}
+
+ io.helidon.common
+ helidon-common-mapper
+ ${project.version}
+
+
+
+
+ io.helidon.dbclient
+ helidon-dbclient
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-common
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-jdbc
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-mongodb
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-mongodb
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-health
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-jsonp
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-metrics
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-metrics-jdbc
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-tracing
+ ${project.version}
+
+
+ io.helidon.dbclient
+ helidon-dbclient-webserver-jsonp
+ ${project.version}
+
+
+
+
+ io.helidon.examples.dbclient
+ helidon-examples-dbclient-common
+ ${project.version}
+
diff --git a/common/configurable/src/main/java/io/helidon/common/configurable/LruCache.java b/common/configurable/src/main/java/io/helidon/common/configurable/LruCache.java
new file mode 100644
index 000000000..6e1503d61
--- /dev/null
+++ b/common/configurable/src/main/java/io/helidon/common/configurable/LruCache.java
@@ -0,0 +1,249 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.common.configurable;
+
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
+
+import io.helidon.config.Config;
+
+/**
+ * Least recently used cache.
+ * This cache has a capacity. When the capacity is reached, the oldest record is removed from the cache when a new one
+ * is added.
+ *
+ * @param type of the keys of the map
+ * @param type of the values of the map
+ */
+public final class LruCache {
+ /**
+ * Default capacity of the cache: {@value}.
+ */
+ public static final int DEFAULT_CAPACITY = 10000;
+
+ private final LinkedHashMap backingMap = new LinkedHashMap<>();
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock(true);
+ private final Lock readLock = rwLock.readLock();
+ private final Lock writeLock = rwLock.writeLock();
+
+ private final int capacity;
+
+ private LruCache(Builder builder) {
+ this.capacity = builder.capacity;
+ }
+
+ /**
+ * Create a new builder.
+ *
+ * @param key type
+ * @param value type
+ * @return a new fluent API builder instance
+ */
+ public static Builder builder() {
+ return new Builder<>();
+ }
+
+ /**
+ * Create an instance with default configuration.
+ *
+ * @param key type
+ * @param value type
+ * @return a new cache instance
+ * @see #DEFAULT_CAPACITY
+ */
+ public static LruCache create() {
+ Builder builder = builder();
+ return builder.build();
+ }
+
+ /**
+ * Get a value from the cache.
+ *
+ * @param key key to retrieve
+ * @return value if present or empty
+ */
+ public Optional get(K key) {
+ readLock.lock();
+
+ V value;
+ try {
+ value = backingMap.get(key);
+ } finally {
+ readLock.unlock();
+ }
+
+ if (null == value) {
+ return Optional.empty();
+ }
+
+ writeLock.lock();
+ try {
+ // make sure the value is the last in the map (I do ignore a race here, as it is not significant)
+ // if some other thread moved another record to the front, we just move ours before it
+
+ // TODO this hurts - we just need to move the key to the last position
+ // maybe this should be replaced with a list and a map?
+ value = backingMap.get(key);
+ if (null == value) {
+ return Optional.empty();
+ }
+ backingMap.remove(key);
+ backingMap.put(key, value);
+
+ return Optional.of(value);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ /**
+ * Remove a value from the cache.
+ *
+ * @param key key of the record to remove
+ * @return the value that was mapped to the key, or empty if none was
+ */
+ public Optional remove(K key) {
+
+ writeLock.lock();
+ try {
+ return Optional.ofNullable(backingMap.remove(key));
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ /**
+ * Put a value to the cache.
+ *
+ * @param key key to add
+ * @param value value to add
+ * @return value that was already mapped or empty if the value was not mapped
+ */
+ public Optional put(K key, V value) {
+ writeLock.lock();
+ try {
+ V currentValue = backingMap.remove(key);
+ if (null == currentValue) {
+ // need to free space - we did not make the map smaller
+ if (backingMap.size() >= capacity) {
+ Iterator iterator = backingMap.values().iterator();
+ iterator.next();
+ iterator.remove();
+ }
+ }
+
+ backingMap.put(key, value);
+ return Optional.ofNullable(currentValue);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ /**
+ * Either return a cached value or compute it and cache it.
+ * In case this method is called in parallel for the same key, the value actually present in the map may be from
+ * any of the calls.
+ * This method always returns either the existing value from the map, or the value provided by the supplier. It
+ * never returns a result from another thread's supplier.
+ *
+ * @param key key to check/insert value for
+ * @param valueSupplier supplier called if the value is not yet cached, or is invalid
+ * @return current value from the cache, or computed value from the supplier
+ */
+ public Optional computeValue(K key, Supplier> valueSupplier) {
+ // get is properly synchronized
+ Optional currentValue = get(key);
+ if (currentValue.isPresent()) {
+ return currentValue;
+ }
+ Optional newValue = valueSupplier.get();
+ // put is also properly synchronized - nevertheless we may replace the value more then once
+ // if called from parallel threads
+ newValue.ifPresent(theValue -> put(key, theValue));
+
+ return newValue;
+ }
+
+ /**
+ * Current size of the map.
+ *
+ * @return number of records currently cached
+ */
+ public int size() {
+ readLock.lock();
+ try {
+ return backingMap.size();
+ } finally {
+ readLock.unlock();
+ }
+ }
+
+ /**
+ * Capacity of this cache.
+ *
+ * @return configured capacity of this cache
+ */
+ public int capacity() {
+ return capacity;
+ }
+
+ // for unit testing
+ V directGet(K key) {
+ return backingMap.get(key);
+ }
+
+ /**
+ * Fluent API builder for {@link io.helidon.common.configurable.LruCache}.
+ *
+ * @param type of keys
+ * @param type of values
+ */
+ public static class Builder implements io.helidon.common.Builder> {
+ private int capacity = DEFAULT_CAPACITY;
+
+ @Override
+ public LruCache build() {
+ return new LruCache<>(this);
+ }
+
+ /**
+ * Load configuration of this cache from configuration.
+ *
+ * @param config configuration
+ * @return updated builder instance
+ */
+ public Builder config(Config config) {
+ config.get("capacity").asInt().ifPresent(this::capacity);
+ return this;
+ }
+
+ /**
+ * Configure capacity of the cache.
+ *
+ * @param capacity maximal number of records in the cache before the oldest one is removed
+ * @return updated builder instance
+ */
+ public Builder capacity(int capacity) {
+ this.capacity = capacity;
+ return this;
+ }
+ }
+}
diff --git a/common/configurable/src/test/java/io/helidon/common/configurable/LruCacheTest.java b/common/configurable/src/test/java/io/helidon/common/configurable/LruCacheTest.java
new file mode 100644
index 000000000..7c8167fc8
--- /dev/null
+++ b/common/configurable/src/test/java/io/helidon/common/configurable/LruCacheTest.java
@@ -0,0 +1,112 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.helidon.common.configurable;
+
+import java.util.Optional;
+
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+/**
+ * Unit test for {@link LruCache}.
+ */
+class LruCacheTest {
+ @Test
+ void testCache() {
+ LruCache theCache = LruCache.create();
+ String value = "cached";
+ String key = "theKey";
+ String newValue = "not-cached";
+
+ Optional res = theCache.put(key, value);
+ assertThat(res, is(Optional.empty()));
+ res = theCache.get(key);
+ assertThat(res, is(Optional.of(value)));
+ res = theCache.computeValue(key, () -> Optional.of(newValue));
+ assertThat(res, is(Optional.of(value)));
+ res = theCache.remove(key);
+ assertThat(res, is(Optional.of(value)));
+ res = theCache.get(key);
+ assertThat(res, is(Optional.empty()));
+ }
+
+ @Test
+ void testCacheComputeValue() {
+ LruCache theCache = LruCache.create();
+ String value = "cached";
+ String key = "theKey";
+ String newValue = "not-cached";
+
+ Optional res = theCache.computeValue(key, () -> Optional.of(value));
+ assertThat(res, is(Optional.of(value)));
+ res = theCache.get(key);
+ assertThat(res, is(Optional.of(value)));
+ res = theCache.computeValue(key, () -> Optional.of(newValue));
+ assertThat(res, is(Optional.of(value)));
+ res = theCache.remove(key);
+ assertThat(res, is(Optional.of(value)));
+ res = theCache.get(key);
+ assertThat(res, is(Optional.empty()));
+ }
+
+ @Test
+ void testMaxCapacity() {
+ LruCache theCache = LruCache.builder().capacity(10).build();
+ for (int i = 0; i < 10; i++) {
+ theCache.put(i, i);
+ }
+ for (int i = 0; i < 10; i++) {
+ Optional integer = theCache.get(i);
+ assertThat(integer, is(Optional.of(i)));
+ }
+ theCache.put(10, 10);
+ Optional res = theCache.get(0);
+ assertThat(res, is(Optional.empty()));
+ res = theCache.get(10);
+ assertThat(res, is(Optional.of(10)));
+ }
+
+ @Test
+ void testLruBehavior() {
+ LruCache theCache = LruCache.builder().capacity(10).build();
+ for (int i = 0; i < 10; i++) {
+ // insert all
+ theCache.put(i, i);
+ }
+ for (int i = 0; i < 10; i++) {
+ // use them in ascending order
+ Optional integer = theCache.get(i);
+ assertThat(integer, is(Optional.of(i)));
+ }
+ // now use 0
+ Optional value = theCache.get(0);
+ assertThat(value, is(Optional.of(0)));
+
+ theCache.put(10, 10);
+
+ // 0 should be in
+ value = theCache.get(0);
+ assertThat(value, is(Optional.of(0)));
+
+ // 1 should not
+ value = theCache.get(1);
+ assertThat(value, is(Optional.empty()));
+
+ }
+}
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/Flows.java b/common/reactive/src/main/java/io/helidon/common/reactive/Flows.java
new file mode 100644
index 000000000..d4e9e87af
--- /dev/null
+++ b/common/reactive/src/main/java/io/helidon/common/reactive/Flows.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.common.reactive;
+
+import java.util.concurrent.Flow;
+
+/**
+ * Utilities for Flow API.
+ */
+public final class Flows {
+ private Flows() {
+ }
+
+ /**
+ * Empty publisher.
+ *
+ * @param type of the publisher
+ * @return a new empty publisher that just completes the subscriber
+ */
+ public static Flow.Publisher emptyPublisher() {
+ return subscriber -> subscriber.onSubscribe(new Flow.Subscription() {
+ @Override
+ public void request(long n) {
+ subscriber.onComplete();
+ }
+
+ @Override
+ public void cancel() {
+ }
+ });
+ }
+
+ /**
+ * A publisher of a single value.
+ *
+ * @param value value to publish
+ * @param type of the publisher
+ * @return a new publisher that publishes the single value and completes the subscriber
+ */
+ public static Flow.Publisher singletonPublisher(T value) {
+ return subscriber -> subscriber.onSubscribe(new Flow.Subscription() {
+ @Override
+ public void request(long n) {
+ subscriber.onNext(value);
+ subscriber.onComplete();
+ }
+
+ @Override
+ public void cancel() {
+ }
+ });
+ }
+}
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/MappingProcessor.java b/common/reactive/src/main/java/io/helidon/common/reactive/MappingProcessor.java
new file mode 100644
index 000000000..435083f1c
--- /dev/null
+++ b/common/reactive/src/main/java/io/helidon/common/reactive/MappingProcessor.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.common.reactive;
+
+import java.util.concurrent.Flow;
+import java.util.function.Function;
+
+/**
+ * A {@link Flow.Processor} that only maps the source type to target type using a mapping function.
+ *
+ * @param type of the publisher we subscribe to
+ * @param type of the publisher we expose
+ */
+public final class MappingProcessor implements Flow.Processor {
+ private final Function resultMapper;
+ private Flow.Subscriber mySubscriber;
+ private Flow.Subscription subscription;
+
+ private MappingProcessor(Function resultMapper) {
+ this.resultMapper = resultMapper;
+ }
+
+ /**
+ * Create a mapping processor for a mapping function.
+ * @param mappingFunction function that maps source to target (applied for each record)
+ * @param Source type
+ * @param Target type
+ * @return a new mapping processor
+ */
+ public static MappingProcessor create(Function mappingFunction) {
+ return new MappingProcessor<>(mappingFunction);
+ }
+
+ @Override
+ public void subscribe(Flow.Subscriber subscriber) {
+ this.mySubscriber = subscriber;
+ subscriber.onSubscribe(new Flow.Subscription() {
+ @Override
+ public void request(long n) {
+ if (null != subscription) {
+ subscription.request(n);
+ }
+ }
+
+ @Override
+ public void cancel() {
+ if (null != subscription) {
+ subscription.cancel();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ this.subscription = subscription;
+ }
+
+ @Override
+ public void onNext(SOURCE item) {
+ mySubscriber.onNext(resultMapper.apply(item));
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ mySubscriber.onError(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ mySubscriber.onComplete();
+ }
+}
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/OptionalCompletionStage.java b/common/reactive/src/main/java/io/helidon/common/reactive/OptionalCompletionStage.java
new file mode 100644
index 000000000..b8785ce39
--- /dev/null
+++ b/common/reactive/src/main/java/io/helidon/common/reactive/OptionalCompletionStage.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.common.reactive;
+
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Consumer;
+
+/**
+ * A completion stage that allows processing of cases when the element
+ * is present and when not.
+ *
+ * @param return type of the asynchronous operation
+ */
+public interface OptionalCompletionStage extends CompletionStage> {
+
+ /**
+ * Returns a new {@code OptionalCompletionStage} that, when this stage completes
+ * normally and returns {@code null}, executes the given action.
+ *
+ * @param action the action to perform before completing the
+ * returned {@code OptionalCompletionStage}
+ * @return the new {@code OptionalCompletionStage}
+ */
+ OptionalCompletionStage onEmpty(Runnable action);
+
+ /**
+ * Returns a new {@code OptionalCompletionStage} that, when this stage completes
+ * normally and returns non-{@code null}, is executed with this stage's
+ * result as the argument to the supplied action.
+ *
+ * @param action the action to perform before completing the
+ * returned {@code OptionalCompletionStage}
+ * @return the new {@code OptionalCompletionStage}
+ */
+ OptionalCompletionStage onValue(Consumer action);
+
+ /**
+ * Creates a new instance of the completion stage that allows processing of cases when the element
+ * is present and when not.
+ *
+ * @param return type of the asynchronous operation
+ * @param originalStage source completion stage instance
+ * @return the new {@code OptionalCompletionStage}
+ */
+ static OptionalCompletionStage create(CompletionStage> originalStage) {
+ return new OptionalCompletionStageImpl<>(originalStage);
+ }
+
+}
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/OptionalCompletionStageImpl.java b/common/reactive/src/main/java/io/helidon/common/reactive/OptionalCompletionStageImpl.java
new file mode 100644
index 000000000..9832fc1ca
--- /dev/null
+++ b/common/reactive/src/main/java/io/helidon/common/reactive/OptionalCompletionStageImpl.java
@@ -0,0 +1,272 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.common.reactive;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * Internal implementation of the completion stage that allows processing of cases when the element is present and when
+ * not.
+ *
+ * @param return type of the asynchronous operation
+ */
+class OptionalCompletionStageImpl implements OptionalCompletionStage {
+
+ private final CompletionStage> originalStage;
+
+ OptionalCompletionStageImpl(CompletionStage> originalStag) {
+ this.originalStage = originalStag;
+ }
+
+ @Override
+ public OptionalCompletionStage onEmpty(Runnable action) {
+ originalStage.thenAccept(original -> {
+ if (!original.isPresent()) {
+ action.run();
+ }
+ });
+ return this;
+ }
+
+ @Override
+ public OptionalCompletionStage onValue(Consumer action) {
+ originalStage.thenAccept(original -> {
+ original.ifPresent(action);
+ });
+ return this;
+ }
+
+ @Override
+ public CompletionStage thenApply(Function, ? extends U> fn) {
+ return originalStage.thenApply(fn);
+ }
+
+ @Override
+ public CompletionStage thenApplyAsync(Function, ? extends U> fn) {
+ return originalStage.thenApplyAsync(fn);
+ }
+
+ @Override
+ public CompletionStage thenApplyAsync(Function, ? extends U> fn,
+ Executor executor) {
+ return originalStage.thenApplyAsync(fn, executor);
+ }
+
+ @Override
+ public CompletionStage thenAccept(Consumer> action) {
+ return originalStage.thenAccept(action);
+ }
+
+ @Override
+ public CompletionStage thenAcceptAsync(Consumer> action) {
+ return originalStage.thenAcceptAsync(action);
+ }
+
+ @Override
+ public CompletionStage thenAcceptAsync(Consumer> action,
+ Executor executor) {
+ return originalStage.thenAcceptAsync(action, executor);
+ }
+
+ @Override
+ public CompletionStage thenRun(Runnable action) {
+ return originalStage.thenRun(action);
+ }
+
+ @Override
+ public CompletionStage thenRunAsync(Runnable action) {
+ return originalStage.thenRunAsync(action);
+ }
+
+ @Override
+ public CompletionStage thenRunAsync(Runnable action, Executor executor) {
+ return originalStage.thenRunAsync(action, executor);
+ }
+
+ @Override
+ public CompletionStage thenCombine(CompletionStage other,
+ BiFunction, ? super U, ? extends V> fn) {
+ return originalStage.thenCombine(other, fn);
+ }
+
+ @Override
+ public CompletionStage thenCombineAsync(CompletionStage other,
+ BiFunction, ? super U, ? extends V> fn) {
+ return originalStage.thenCombineAsync(other, fn);
+ }
+
+ @Override
+ public CompletionStage thenCombineAsync(CompletionStage other,
+ BiFunction, ? super U, ? extends V> fn,
+ Executor executor) {
+ return originalStage.thenCombineAsync(other, fn, executor);
+ }
+
+ @Override
+ public CompletionStage thenAcceptBoth(CompletionStage other,
+ BiConsumer, ? super U> action) {
+ return originalStage.thenAcceptBoth(other, action);
+ }
+
+ @Override
+ public CompletionStage thenAcceptBothAsync(CompletionStage other,
+ BiConsumer, ? super U> action) {
+ return originalStage.thenAcceptBothAsync(other, action);
+ }
+
+ @Override
+ public CompletionStage thenAcceptBothAsync(CompletionStage other,
+ BiConsumer, ? super U> action,
+ Executor executor) {
+ return originalStage.thenAcceptBothAsync(other, action, executor);
+ }
+
+ @Override
+ public CompletionStage runAfterBoth(CompletionStage other, Runnable action) {
+ return originalStage.runAfterBoth(other, action);
+ }
+
+ @Override
+ public CompletionStage runAfterBothAsync(CompletionStage other, Runnable action) {
+ return originalStage.runAfterBothAsync(other, action);
+ }
+
+ @Override
+ public CompletionStage runAfterBothAsync(CompletionStage other, Runnable action, Executor executor) {
+ return originalStage.runAfterBothAsync(other, action, executor);
+ }
+
+ @Override
+ public CompletionStage applyToEither(CompletionStage> other,
+ Function, U> fn) {
+ return originalStage.applyToEither(other, fn);
+ }
+
+ @Override
+ public CompletionStage applyToEitherAsync(CompletionStage> other,
+ Function, U> fn) {
+ return originalStage.applyToEitherAsync(other, fn);
+ }
+
+ @Override
+ public CompletionStage applyToEitherAsync(CompletionStage> other,
+ Function, U> fn,
+ Executor executor) {
+ return originalStage.applyToEitherAsync(other, fn, executor);
+ }
+
+ @Override
+ public CompletionStage acceptEither(CompletionStage> other,
+ Consumer> action) {
+ return originalStage.acceptEither(other, action);
+ }
+
+ @Override
+ public CompletionStage acceptEitherAsync(CompletionStage> other,
+ Consumer> action) {
+ return originalStage.acceptEitherAsync(other, action);
+ }
+
+ @Override
+ public CompletionStage acceptEitherAsync(CompletionStage> other,
+ Consumer> action,
+ Executor executor) {
+ return originalStage.acceptEitherAsync(other, action, executor);
+ }
+
+ @Override
+ public CompletionStage runAfterEither(CompletionStage other, Runnable action) {
+ return originalStage.runAfterEither(other, action);
+ }
+
+ @Override
+ public CompletionStage runAfterEitherAsync(CompletionStage other, Runnable action) {
+ return originalStage.runAfterEitherAsync(other, action);
+ }
+
+ @Override
+ public CompletionStage runAfterEitherAsync(CompletionStage other,
+ Runnable action,
+ Executor executor) {
+ return originalStage.runAfterEitherAsync(other, action, executor);
+ }
+
+ @Override
+ public CompletionStage thenCompose(Function, ? extends CompletionStage> fn) {
+ return originalStage.thenCompose(fn);
+ }
+
+ @Override
+ public CompletionStage thenComposeAsync(Function, ? extends CompletionStage> fn) {
+ return originalStage.thenComposeAsync(fn);
+ }
+
+ @Override
+ public CompletionStage thenComposeAsync(Function, ? extends CompletionStage> fn,
+ Executor executor) {
+ return originalStage.thenComposeAsync(fn, executor);
+ }
+
+ @Override
+ public CompletionStage> exceptionally(Function> fn) {
+ return originalStage.exceptionally(fn);
+ }
+
+ @Override
+ public CompletionStage> whenComplete(BiConsumer, ? super Throwable> action) {
+ return originalStage.whenComplete(action);
+ }
+
+ @Override
+ public CompletionStage> whenCompleteAsync(BiConsumer, ? super Throwable> action) {
+ return originalStage.whenCompleteAsync(action);
+ }
+
+ @Override
+ public CompletionStage> whenCompleteAsync(BiConsumer, ? super Throwable> action,
+ Executor executor) {
+ return originalStage.whenCompleteAsync(action, executor);
+ }
+
+ @Override
+ public CompletionStage handle(BiFunction, Throwable, ? extends U> fn) {
+ return originalStage.handle(fn);
+ }
+
+ @Override
+ public CompletionStage handleAsync(BiFunction, Throwable, ? extends U> fn) {
+ return originalStage.handleAsync(fn);
+ }
+
+ @Override
+ public CompletionStage handleAsync(BiFunction, Throwable, ? extends U> fn,
+ Executor executor) {
+ return originalStage.handleAsync(fn, executor);
+ }
+
+ @Override
+ public CompletableFuture> toCompletableFuture() {
+ return originalStage.toCompletableFuture();
+ }
+
+}
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/Single.java b/common/reactive/src/main/java/io/helidon/common/reactive/Single.java
index 86070daed..8d7e8f9b0 100644
--- a/common/reactive/src/main/java/io/helidon/common/reactive/Single.java
+++ b/common/reactive/src/main/java/io/helidon/common/reactive/Single.java
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
package io.helidon.common.reactive;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
@@ -62,8 +63,9 @@ public interface Single extends Subscribable {
}
/**
- * Exposes this {@link Single} instance as a {@link CompletionStage}. Note that if this {@link Single} completes without a
- * value, the resulting {@link CompletionStage} will be completed exceptionally with an {@link IllegalStateException}
+ * Exposes this {@link Single} instance as a {@link CompletionStage}.
+ * Note that if this {@link Single} completes without a value, the resulting {@link CompletionStage} will be completed
+ * exceptionally with an {@link IllegalStateException}
*
* @return CompletionStage
*/
@@ -79,6 +81,26 @@ public interface Single extends Subscribable {
}
}
+ /**
+ * Exposes this {@link Single} instance as a {@link CompletionStage} with {@code Optional} return type
+ * of the asynchronous operation.
+ * Note that if this {@link Single} completes without a value, the resulting {@link CompletionStage} will be completed
+ * exceptionally with an {@link IllegalStateException}
+ *
+ * @return CompletionStage
+ */
+ default CompletionStage> toOptionalStage() {
+ try {
+ SingleToOptionalFuture subscriber = new SingleToOptionalFuture<>();
+ this.subscribe(subscriber);
+ return subscriber;
+ } catch (Throwable ex) {
+ CompletableFuture> future = new CompletableFuture<>();
+ future.completeExceptionally(ex);
+ return future;
+ }
+ }
+
/**
* Short-hand for {@code toFuture().toCompletableFuture().get()}.
* @return T
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/SingleToOptionalFuture.java b/common/reactive/src/main/java/io/helidon/common/reactive/SingleToOptionalFuture.java
new file mode 100644
index 000000000..310add45c
--- /dev/null
+++ b/common/reactive/src/main/java/io/helidon/common/reactive/SingleToOptionalFuture.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.common.reactive;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Subscriber;
+import java.util.concurrent.Flow.Subscription;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * {@link io.helidon.common.reactive.Single} exposed as a {@link java.util.concurrent.CompletableFuture}.
+ */
+class SingleToOptionalFuture extends CompletableFuture> implements Subscriber {
+
+ private final AtomicReference ref = new AtomicReference<>();
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ boolean cancelled = super.cancel(mayInterruptIfRunning);
+ if (cancelled) {
+ Subscription s = ref.getAndSet(null);
+ if (s != null) {
+ s.cancel();
+ }
+ }
+ return cancelled;
+ }
+
+ @Override
+ public void onSubscribe(Subscription next) {
+ Subscription current = ref.getAndSet(next);
+ Objects.requireNonNull(next, "Subscription cannot be null");
+ if (current != null) {
+ next.cancel();
+ current.cancel();
+ } else {
+ next.request(Long.MAX_VALUE);
+ }
+ }
+
+ @Override
+ public void onNext(T item) {
+ Subscription s = ref.getAndSet(null);
+ if (s != null) {
+ super.complete(Optional.ofNullable(item));
+ s.cancel();
+ }
+ }
+
+ @Override
+ public void onError(Throwable ex) {
+ if (ref.getAndSet(null) != null) {
+ super.completeExceptionally(ex);
+ }
+ }
+
+ @Override
+ public void onComplete() {
+ if (ref.getAndSet(null) != null) {
+ super.complete(Optional.empty());
+ }
+ }
+
+ @Override
+ public boolean complete(Optional value) {
+ throw new UnsupportedOperationException("This future cannot be completed manually");
+ }
+
+ @Override
+ public boolean completeExceptionally(Throwable ex) {
+ throw new UnsupportedOperationException("This future cannot be completed manually");
+ }
+}
diff --git a/config/config/src/test/java/io/helidon/config/ConfigSupplierTest.java b/config/config/src/test/java/io/helidon/config/ConfigSupplierTest.java
index e78f50d76..0ffb61baf 100644
--- a/config/config/src/test/java/io/helidon/config/ConfigSupplierTest.java
+++ b/config/config/src/test/java/io/helidon/config/ConfigSupplierTest.java
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2020 Oracle and/or its affiliates. All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@ import io.helidon.config.spi.ConfigNode;
import io.helidon.config.spi.ConfigNode.ObjectNode;
import io.helidon.config.spi.TestingConfigSource;
+import org.junit.Ignore;
import org.junit.jupiter.api.Test;
import static io.helidon.config.ConfigTest.waitForAssert;
@@ -175,7 +176,13 @@ public class ConfigSupplierTest {
is(ConfigValues.simpleValue("NEW item 1")));
}
+ @Ignore
@Test
+ // TODO cause of intermittent test failures:
+ /*
+ Tests in error:
+ ConfigSupplierTest.testSupplierFromMissingToListNode:209->lambda$testSupplierFromMissingToListNode$16:216 ยป IllegalState
+ */
public void testSupplierFromMissingToListNode() throws InterruptedException {
// config source
TestingConfigSource configSource = TestingConfigSource.builder().build();
diff --git a/dbclient/common/pom.xml b/dbclient/common/pom.xml
new file mode 100644
index 000000000..5caf7abd7
--- /dev/null
+++ b/dbclient/common/pom.xml
@@ -0,0 +1,51 @@
+
+
+
+
+
+ helidon-dbclient-project
+ io.helidon.dbclient
+ 2.0-SNAPSHOT
+
+ 4.0.0
+
+ helidon-dbclient-common
+ Helidon DB Client Common
+
+
+
+ io.helidon.dbclient
+ helidon-dbclient
+
+
+ io.helidon.common
+ helidon-common-configurable
+
+
+ org.junit.jupiter
+ junit-jupiter-api
+ test
+
+
+ org.hamcrest
+ hamcrest-all
+ test
+
+
+
diff --git a/dbclient/common/src/main/java/io/helidon/dbclient/common/AbstractDbExecute.java b/dbclient/common/src/main/java/io/helidon/dbclient/common/AbstractDbExecute.java
new file mode 100644
index 000000000..a78dfeab1
--- /dev/null
+++ b/dbclient/common/src/main/java/io/helidon/dbclient/common/AbstractDbExecute.java
@@ -0,0 +1,149 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.dbclient.common;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
+
+import io.helidon.dbclient.DbExecute;
+import io.helidon.dbclient.DbStatementDml;
+import io.helidon.dbclient.DbStatementGeneric;
+import io.helidon.dbclient.DbStatementGet;
+import io.helidon.dbclient.DbStatementQuery;
+import io.helidon.dbclient.DbStatementType;
+import io.helidon.dbclient.DbStatements;
+
+/**
+ * Implements methods that do not require implementation for each provider.
+ */
+public abstract class AbstractDbExecute implements DbExecute {
+ private final DbStatements statements;
+
+ /**
+ * Create an instance with configured statements.
+ *
+ * @param statements statements to obtains named statements, esp. for {@link #statementText(String)}.
+ */
+ protected AbstractDbExecute(DbStatements statements) {
+ this.statements = statements;
+ }
+
+ /**
+ * Return a statement text based on the statement name.
+ * This is a utility method that probably would use {@link io.helidon.dbclient.DbStatements} to retrieve the named statements.
+ *
+ * @param name name of the statement
+ * @return statement text
+ */
+ protected String statementText(String name) {
+ return statements.statement(name);
+ }
+
+ @Override
+ public DbStatementQuery createNamedQuery(String statementName) {
+ return createNamedQuery(statementName, statementText(statementName));
+ }
+
+ @Override
+ public DbStatementQuery createQuery(String statement) {
+ return createNamedQuery(generateName(DbStatementType.QUERY, statement), statement);
+ }
+
+ @Override
+ public DbStatementGet createNamedGet(String statementName) {
+ return createNamedGet(statementName, statementText(statementName));
+ }
+
+ @Override
+ public DbStatementGet createGet(String statement) {
+ return createNamedGet(generateName(DbStatementType.GET, statement), statement);
+ }
+
+ @Override
+ public DbStatementDml createNamedInsert(String statementName) {
+ return createNamedInsert(statementName, statementText(statementName));
+ }
+
+ @Override
+ public DbStatementDml createInsert(String statement) {
+ return createNamedInsert(generateName(DbStatementType.INSERT, statement), statement);
+ }
+
+ @Override
+ public DbStatementDml createNamedUpdate(String statementName) {
+ return createNamedUpdate(statementName, statementText(statementName));
+ }
+
+ @Override
+ public DbStatementDml createUpdate(String statement) {
+ return createNamedUpdate(generateName(DbStatementType.UPDATE, statement), statement);
+ }
+
+ @Override
+ public DbStatementDml createNamedDelete(String statementName) {
+ return createNamedDelete(statementName, statementText(statementName));
+ }
+
+ @Override
+ public DbStatementDml createDelete(String statement) {
+ return createNamedDelete(generateName(DbStatementType.DELETE, statement), statement);
+ }
+
+ @Override
+ public DbStatementDml createNamedDmlStatement(String statementName) {
+ return createNamedDmlStatement(statementName, statementText(statementName));
+ }
+
+ @Override
+ public DbStatementDml createDmlStatement(String statement) {
+ return createNamedDmlStatement(generateName(DbStatementType.DML, statement), statement);
+ }
+
+ @Override
+ public DbStatementGeneric createNamedStatement(String statementName) {
+ return createNamedStatement(statementName, statementText(statementName));
+ }
+
+ @Override
+ public DbStatementGeneric createStatement(String statement) {
+ return createNamedStatement(generateName(DbStatementType.UNKNOWN, statement), statement);
+ }
+
+ /**
+ * Generate a name for a statement.
+ * The default implementation uses {@code SHA-256} so the same name is always
+ * returned for the same statement.
+ *
+ * As there is always a small risk of duplicity, named statements are recommended!
+ *
+ * @param type type of the statement
+ * @param statement statement that it going to be executed
+ * @return name of the statement
+ */
+ protected String generateName(DbStatementType type, String statement) {
+ String sha256;
+ try {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ digest.update(statement.getBytes(StandardCharsets.UTF_8));
+ sha256 = Base64.getEncoder().encodeToString(digest.digest());
+ } catch (NoSuchAlgorithmException ignored) {
+ return "sha256failed";
+ }
+ return type.prefix() + '_' + sha256;
+ }
+}
diff --git a/dbclient/common/src/main/java/io/helidon/dbclient/common/AbstractStatement.java b/dbclient/common/src/main/java/io/helidon/dbclient/common/AbstractStatement.java
new file mode 100644
index 000000000..2d67608a6
--- /dev/null
+++ b/dbclient/common/src/main/java/io/helidon/dbclient/common/AbstractStatement.java
@@ -0,0 +1,296 @@
+/*
+ * Copyright (c) 2019, 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.dbclient.common;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.logging.Logger;
+
+import io.helidon.common.context.Context;
+import io.helidon.common.context.Contexts;
+import io.helidon.common.mapper.MapperManager;
+import io.helidon.dbclient.DbInterceptor;
+import io.helidon.dbclient.DbInterceptorContext;
+import io.helidon.dbclient.DbMapperManager;
+import io.helidon.dbclient.DbStatement;
+import io.helidon.dbclient.DbStatementType;
+
+/**
+ * Common statement methods and fields.
+ *
+ * @param type of a subclass
+ * @param the result type of the statement as returned by {@link #execute()}
+ */
+public abstract class AbstractStatement, R> implements DbStatement {
+
+ /** Local logger instance. */
+ private static final Logger LOGGER = Logger.getLogger(AbstractStatement.class.getName());
+
+ private ParamType paramType = ParamType.UNKNOWN;
+ private StatementParameters parameters;
+ private final DbStatementType dbStatementType;
+ private final String statementName;
+ private final String statement;
+ private final DbMapperManager dbMapperManager;
+ private final MapperManager mapperManager;
+ private final InterceptorSupport interceptors;
+
+ /**
+ * Statement that handles parameters.
+ *
+ * @param dbStatementType type of this statement
+ * @param statementName name of this statement
+ * @param statement text of this statement
+ * @param dbMapperManager db mapper manager to use when mapping types to parameters
+ * @param mapperManager mapper manager to use when mapping results
+ * @param interceptors interceptors to be executed
+ */
+ protected AbstractStatement(DbStatementType dbStatementType,
+ String statementName,
+ String statement,
+ DbMapperManager dbMapperManager,
+ MapperManager mapperManager,
+ InterceptorSupport interceptors) {
+ this.dbStatementType = dbStatementType;
+ this.statementName = statementName;
+ this.statement = statement;
+ this.dbMapperManager = dbMapperManager;
+ this.mapperManager = mapperManager;
+ this.interceptors = interceptors;
+ }
+
+ @Override
+ public CompletionStage execute() {
+ CompletableFuture queryFuture = new CompletableFuture<>();
+ CompletableFuture statementFuture = new CompletableFuture<>();
+ DbInterceptorContext dbContext = DbInterceptorContext.create(dbType())
+ .resultFuture(queryFuture)
+ .statementFuture(statementFuture);
+
+ update(dbContext);
+ CompletionStage dbContextFuture = invokeInterceptors(dbContext);
+ return doExecute(dbContextFuture, statementFuture, queryFuture);
+ }
+
+ /**
+ * Invoke all interceptors.
+ *
+ * @param dbContext initial interceptor context
+ * @return future with the result of interceptors processing
+ */
+ CompletionStage invokeInterceptors(DbInterceptorContext dbContext) {
+ CompletableFuture result = CompletableFuture.completedFuture(dbContext);
+
+ dbContext.context(Contexts.context().orElseGet(Context::create));
+
+ for (DbInterceptor interceptor : interceptors.interceptors(statementType(), statementName())) {
+ result = result.thenCompose(interceptor::statement);
+ }
+
+ return result;
+ }
+
+ /**
+ * Type of this statement.
+ *
+ * @return statement type
+ */
+ protected DbStatementType statementType() {
+ return dbStatementType;
+ }
+
+ /**
+ * Execute the statement against the database.
+ *
+ * @param dbContext future that completes after all interceptors are invoked
+ * @param statementFuture future that should complete when the statement finishes execution
+ * @param queryFuture future that should complete when the result set is fully read (if one exists),
+ * otherwise complete same as statementFuture
+ * @return result of this db statement.
+ */
+ protected abstract CompletionStage doExecute(CompletionStage dbContext,
+ CompletableFuture statementFuture,
+ CompletableFuture queryFuture);
+
+ /**
+ * Type of this database to use in interceptor context.
+ *
+ * @return type of this db
+ */
+ protected abstract String dbType();
+
+ @Override
+ public S params(List parameters) {
+ Objects.requireNonNull(parameters, "Parameters cannot be null (may be an empty list)");
+
+ initParameters(ParamType.INDEXED);
+ this.parameters.params(parameters);
+
+ return me();
+ }
+
+ @Override
+ public S params(Map parameters) {
+ initParameters(ParamType.NAMED);
+ this.parameters.params(parameters);
+ return me();
+ }
+
+ @Override
+ public S namedParam(Object parameters) {
+ initParameters(ParamType.NAMED);
+ this.parameters.namedParam(parameters);
+ return me();
+ }
+
+ @Override
+ public S indexedParam(Object parameters) {
+ initParameters(ParamType.INDEXED);
+ this.parameters.indexedParam(parameters);
+ return me();
+ }
+
+ @Override
+ public S addParam(Object parameter) {
+ initParameters(ParamType.INDEXED);
+ this.parameters.addParam(parameter);
+ return me();
+ }
+
+ @Override
+ public S addParam(String name, Object parameter) {
+ initParameters(ParamType.NAMED);
+ this.parameters.addParam(name, parameter);
+ return me();
+ }
+
+ /**
+ * Type of parameters of this statement.
+ *
+ * @return indexed or named, or unknown in case it could not be yet defined
+ */
+ protected ParamType paramType() {
+ return paramType;
+ }
+
+ /**
+ * Db mapper manager.
+ *
+ * @return mapper manager for DB types
+ */
+ protected DbMapperManager dbMapperManager() {
+ return dbMapperManager;
+ }
+
+ /**
+ * Mapper manager.
+ *
+ * @return generic mapper manager
+ */
+ protected MapperManager mapperManager() {
+ return mapperManager;
+ }
+
+ /**
+ * Get the named parameters of this statement.
+ *
+ * @return name parameter map
+ * @throws java.lang.IllegalStateException in case this statement is using indexed parameters
+ */
+ protected Map namedParams() {
+ initParameters(ParamType.NAMED);
+ return parameters.namedParams();
+ }
+
+ /**
+ * Get the indexed parameters of this statement.
+ *
+ * @return parameter list
+ * @throws java.lang.IllegalStateException in case this statement is using named parameters
+ */
+ protected List