Issue ReactiveX#68: simple bulkhead operator

* implementing simple bulkhead

* updating unit tests

* updating gradle files

* Update BulkheadUtils.java

* Adding registry unit test.

* SemaphoreBulkhead test update

* updating unit tests

* addressing code quality checks

* Operator coverage bump

* test name updates

* introducing BulkheadConfig

* updating unit tests

* updating documentation

resilience4j-all/build.gradle

@@ -5,6 +5,7 @@ dependencies {
     compile project(':resilience4j-ratelimiter')
     compile project(':resilience4j-circularbuffer')
     compile project(':resilience4j-circuitbreaker')
+    compile project(':resilience4j-bulkhead')
     compile project(':resilience4j-retry')
     compile project(':resilience4j-consumer')
     compile project(':resilience4j-cache')

resilience4j-bulkhead/build.gradle

@@ -0,0 +1,7 @@
+apply plugin: 'jcstress'
+
+dependencies {
+    compile project(':resilience4j-core')
+    compile ( libraries.rxjava2)
+    testCompile project(':resilience4j-test')
+}

resilience4j-bulkhead/src/jcstress/java/io/github/resilience4j/bulkhead/concurrent/ConcurrentBulkheadTest.java

@@ -0,0 +1,86 @@
+/*
+ *
+ *  Copyright 2017 Robert Winkler, Bohdan Storozhuk, Lucas Lech
+ *
+ *  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.github.resilience4j.bulkhead.concurrent;
+
+import io.github.resilience4j.bulkhead.Bulkhead;
+import io.github.resilience4j.bulkhead.BulkheadConfig;
+import io.github.resilience4j.bulkhead.event.BulkheadEvent;
+import io.github.resilience4j.bulkhead.event.BulkheadEvent.Type;
+import io.reactivex.subscribers.TestSubscriber;
+import org.openjdk.jcstress.annotations.*;
+import org.openjdk.jcstress.infra.results.StringResult1;
+
+import java.text.MessageFormat;
+
+
+@JCStressTest
+@State
+@Outcome.Outcomes(
+    {
+        @Outcome(
+            id = "remainingDepth=1" +
+                 " events=\\[\\[CALL_REJECTED\\], \\[\\], \\[\\]\\]",
+            expect = Expect.ACCEPTABLE
+        ),
+        @Outcome(
+            id = "remainingDepth=1" +
+                 " events=\\[\\[\\], \\[\\], \\[\\]\\]",
+            expect = Expect.ACCEPTABLE
+        )
+    }
+)
+public class ConcurrentBulkheadTest {
+
+    private Bulkhead bulkhead;
+    private TestSubscriber<Type> callRejectectedEventSubscriber;
+
+    public ConcurrentBulkheadTest() {
+
+        bulkhead = Bulkhead.of("test", BulkheadConfig.custom().maxConcurrentCalls(1).build());
+
+        callRejectectedEventSubscriber = bulkhead.getEventStream()
+                                                 .filter(event -> event.getEventType() == Type.CALL_REJECTED)
+                                                 .map(BulkheadEvent::getEventType)
+                                                 .test();
+    }
+
+    @Actor
+    public void firstActor() {
+        if (bulkhead.isCallPermitted()) {
+            bulkhead.onComplete();
+        }
+    }
+
+    @Actor
+    public void secondActor() {
+        if (bulkhead.isCallPermitted()) {
+            bulkhead.onComplete();
+        }
+    }
+
+    @Arbiter
+    public void arbiter(StringResult1 result1) {
+        String result = MessageFormat.format(
+            "remainingDepth={0} events={1}",
+            bulkhead.getAvailableConcurrentCalls(),
+            callRejectectedEventSubscriber.getEvents()
+        );
+        result1.r1 = result;
+    }
+}

resilience4j-bulkhead/src/jmh/java/io/github/resilience4j/bulkhead/BulkheadBenchmark.java

@@ -0,0 +1,82 @@
+/*
+ *
+ *  Copyright 2017 Robert Winkler, Lucas Lech
+ *
+ *  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.github.resilience4j.bulkhead;
+
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@BenchmarkMode(Mode.Throughput)
+public class BulkheadBenchmark {
+
+    private static final int ITERATION_COUNT = 10;
+    private static final int WARMUP_COUNT = 10;
+    private static final int THREAD_COUNT = 2;
+    private static final int FORK_COUNT = 2;
+
+    private Supplier<String> protectedSupplier;
+    private Supplier<String> protectedSupplierWithSb;
+    private Supplier<String> stringSupplier;
+
+    @Setup
+    public void setUp() {
+        stringSupplier = () -> {
+            Blackhole.consumeCPU(100);
+            return "Hello Benchmark";
+        };
+
+        Bulkhead bulkhead = Bulkhead.of("test", 2);
+        protectedSupplier = Bulkhead.decorateSupplier(bulkhead, stringSupplier);
+
+        Bulkhead bulkheadWithSubscriber = Bulkhead.of("test-with-subscriber", 2);
+        bulkheadWithSubscriber.getEventStream().subscribe();
+        protectedSupplierWithSb = Bulkhead.decorateSupplier(bulkheadWithSubscriber, stringSupplier);
+    }
+
+    @Benchmark
+    @Fork(value = FORK_COUNT)
+    @Threads(value = THREAD_COUNT)
+    @Warmup(iterations = WARMUP_COUNT)
+    @Measurement(iterations = ITERATION_COUNT)
+    public String directSupplier() {
+        return stringSupplier.get();
+    }
+
+    @Benchmark
+    @Fork(value = FORK_COUNT)
+    @Threads(value = THREAD_COUNT)
+    @Warmup(iterations = WARMUP_COUNT)
+    @Measurement(iterations = ITERATION_COUNT)
+    public String protectedSupplier() {
+        return protectedSupplier.get();
+    }
+
+    @Benchmark
+    @Fork(value = FORK_COUNT)
+    @Threads(value = THREAD_COUNT)
+    @Warmup(iterations = WARMUP_COUNT)
+    @Measurement(iterations = ITERATION_COUNT)
+    public String protectedSupplierWithSubscriber() {
+        return protectedSupplierWithSb.get();
+    }
+}