Issue ReactiveX#42: Added a new EventConsumerRegistry which can be us…

…ed to create and manage instances of EventConsumers.

RELEASENOTES.adoc

@@ -82,4 +82,5 @@
 * Updated rxjava from 2.0.1 to 2.0.3
 
 == Version 0.8.1
-* Issue #41: Added a method to the CircuitBreakerRegistry which returns a list of all managed CircuitBreaker instances
+* Issue #41: Added a method to the CircuitBreakerRegistry which returns a list of all managed CircuitBreaker instances
+* Issue #42: Added a new EventConsumerRegistry which can be used to create and manage instances of EventConsumers.

src/main/java/io/github/robwin/consumer/CircularEventConsumer.java

@@ -20,13 +20,12 @@
 
 import io.github.robwin.circularbuffer.CircularFifoBuffer;
 import io.github.robwin.circularbuffer.ConcurrentCircularFifoBuffer;
-import io.reactivex.functions.Consumer;
 import javaslang.collection.List;
 
 /**
  * A RxJava consumer which stores CircuitBreakerEvents in a circular buffer with a fixed capacity.
  */
-public class CircularEventConsumer<T> implements Consumer<T>{
+public class CircularEventConsumer<T> implements EventConsumer<T>{
 
     private CircularFifoBuffer<T> eventCircularFifoBuffer;
 
@@ -51,6 +50,7 @@ public void accept(T event) throws Exception {
      *
      * @return a list containing all of the buffered events.
      */
+    @Override
     public List<T> getBufferedEvents(){
         return eventCircularFifoBuffer.toList();
     }

src/main/java/io/github/robwin/consumer/DefaultEventConsumerRegistry.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2017 Robert Winkler
+ *
+ * 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.robwin.consumer;
+
+import javaslang.collection.Array;
+import javaslang.collection.Seq;
+
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class DefaultEventConsumerRegistry<T> implements EventConsumerRegistry<T>{
+
+    /**
+     * The CircularEventConsumers, indexed by name of the backend.
+     */
+    private final ConcurrentMap<String, EventConsumer<T>> registry;
+
+    /**
+     * The constructor with default circuitBreaker properties.
+     */
+    public DefaultEventConsumerRegistry() {
+        this.registry = new ConcurrentHashMap<>();
+    }
+
+    @Override
+    public EventConsumer<T> createEventConsumer(String id, int bufferSize) {
+        EventConsumer<T> eventConsumer = new CircularEventConsumer<>(bufferSize);
+        registry.put(id, eventConsumer);
+        return eventConsumer;
+    }
+
+    @Override
+    public EventConsumer<T> getEventConsumer(String id){
+        return registry.get(id);
+    }
+
+    @Override
+    public Seq<EventConsumer<T>> getAllEventConsumer(){
+        return Array.ofAll(registry.values());
+    }
+}

src/main/java/io/github/robwin/consumer/EventConsumer.java

@@ -0,0 +1,32 @@
+/*
+ *
+ *  Copyright 2016 Robert Winkler
+ *
+ *  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.robwin.consumer;
+
+import io.reactivex.functions.Consumer;
+import javaslang.collection.List;
+
+public interface EventConsumer<T> extends Consumer<T> {
+
+    /**
+     * Returns a list containing all of the buffered events.
+     *
+     * @return a list containing all of the buffered events.
+     */
+    List<T> getBufferedEvents();
+}

src/main/java/io/github/robwin/consumer/EventConsumerRegistry.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2017 Robert Winkler
+ *
+ * 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.robwin.consumer;
+
+import javaslang.collection.Seq;
+
+
+public interface EventConsumerRegistry<T> {
+
+    /**
+     * Creates a new EventConsumer and stores the instance in the registry.
+     *
+     * @param id the id of the EventConsumer
+     * @param bufferSize the size of the EventConsumer
+     * @return a new EventConsumer
+     */
+    EventConsumer<T> createEventConsumer(String id, int bufferSize);
+
+    /**
+     * Returns the EventConsumer to which the specified id is mapped.
+     *
+     * @param id the id of the EventConsumer
+     * @return the EventConsumer to which the specified id is mapped
+     */
+    EventConsumer<T> getEventConsumer(String id);
+
+    /**
+     * Returns all EventConsumer instances.
+     *
+     * @return all EventConsumer instances
+     */
+    Seq<EventConsumer<T>> getAllEventConsumer();
+}

src/test/java/io/github/robwin/consumer/EventConsumerRegistryTest.java

@@ -0,0 +1,54 @@
+/*
+ *
+ *  Copyright 2016 Robert Winkler
+ *
+ *  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.robwin.consumer;
+
+import io.github.robwin.circuitbreaker.event.CircuitBreakerEvent;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class EventConsumerRegistryTest {
+
+    @Test
+    public void shouldCreateAnEventConsumer() {
+        EventConsumerRegistry<CircuitBreakerEvent> registry = new DefaultEventConsumerRegistry<>();
+        EventConsumer<CircuitBreakerEvent> eventEventConsumer = registry.createEventConsumer("testName", 5);
+
+        assertThat(eventEventConsumer).isNotNull();
+        assertThat(eventEventConsumer.getBufferedEvents()).hasSize(0);
+    }
+
+    @Test
+    public void shouldReturnTheSameEventConsumer() {
+        EventConsumerRegistry<CircuitBreakerEvent> registry = new DefaultEventConsumerRegistry<>();
+        EventConsumer<CircuitBreakerEvent> eventEventConsumer1 = registry.createEventConsumer("testName", 5);
+        EventConsumer<CircuitBreakerEvent> eventEventConsumer2 = registry.getEventConsumer("testName");
+
+        assertThat(eventEventConsumer1).isEqualTo(eventEventConsumer2);
+    }
+
+    @Test
+    public void shouldReturnAllEventConsumer() {
+        EventConsumerRegistry<CircuitBreakerEvent> registry = new DefaultEventConsumerRegistry<>();
+        registry.createEventConsumer("testName1", 5);
+        registry.createEventConsumer("testName2", 2);
+
+        assertThat(registry.getAllEventConsumer()).hasSize(2);
+    }
+}