Author: csierra
Date: Fri Nov 17 15:59:59 2017
New Revision: 1815578
URL: http://svn.apache.org/viewvc?rev=1815578&view=rev
Log:
[Component-DSL] Replace router with transformer
The router abstraction was powerful, but not enough to handle some
situations, like aggregations, and added complexity.
Added:
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/Transformer.java
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/TransformerOSGi.java
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/HighestRankingTransformer.java
Removed:
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/RouteOsgiImpl.java
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/HighestRankingRouter.java
Modified:
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/OSGi.java
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/OSGiImpl.java
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/ComponentTest.java
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/DSLTest.java
Modified:
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/OSGi.java
URL:
http://svn.apache.org/viewvc/aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/OSGi.java?rev=1815578&r1=1815577&r2=1815578&view=diff
==============================================================================
---
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/OSGi.java
(original)
+++
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/OSGi.java
Fri Nov 17 15:59:59 2017
@@ -81,6 +81,8 @@ public interface OSGi<T> extends OSGiRun
OSGi<Void> foreach(
Consumer<? super T> onAdded, Consumer<? super T> onRemoved);
+ <S> OSGi<S> transformer(Function<Function<S, Runnable>, Function<T,
Runnable>> fun);
+
static OSGi<BundleContext> bundleContext() {
return new BundleContextOSGiImpl();
@@ -196,28 +198,24 @@ public interface OSGi<T> extends OSGiRun
}
public static <T> OSGi<T> once(OSGi<T> program) {
- return program.route(router -> {
+ return program.transformer(op -> {
AtomicInteger count = new AtomicInteger();
- AtomicReference<SentEvent> terminator = new
AtomicReference<>();
-
- router.onIncoming(t -> {
- int c = count.getAndIncrement();
+ AtomicReference<Runnable> terminator = new
AtomicReference<>();
- if (c == 0) {
- terminator.set(router.signalAdd(t));
+ return t -> {
+ if (count.getAndIncrement() == 0) {
+ terminator.set(op.apply(t));
}
- });
-
- router.onLeaving(t -> {
- int c = count.decrementAndGet();
- if (c == 0) {
- SentEvent s =
terminator.getAndSet(null);
-
- s.terminate();
- }
- });
+ return () -> {
+ if (count.decrementAndGet() == 0) {
+ Runnable runnable =
terminator.getAndSet(NOOP);
+
+ runnable.run();
+ }
+ };
+ };
});
}
@@ -266,19 +264,6 @@ public interface OSGi<T> extends OSGiRun
OSGi<T> filter(Predicate<T> predicate);
- OSGi<T> route(Consumer<Router<T>> routerConsumer);
-
- interface Router<T> {
-
- void onIncoming(Consumer<Event<T>> adding);
- void onLeaving(Consumer<Event<T>> removing);
-
- void onStart(Runnable start);
- void onClose(Runnable close);
-
- SentEvent<T> signalAdd(Event<T> event);
- }
-
public default <S> OSGi<S> applyTo(OSGi<Function<T, S>> fun) {
return fun.flatMap(this::map);
}
Added:
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/Transformer.java
URL:
http://svn.apache.org/viewvc/aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/Transformer.java?rev=1815578&view=auto
==============================================================================
---
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/Transformer.java
(added)
+++
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/Transformer.java
Fri Nov 17 15:59:59 2017
@@ -0,0 +1,10 @@
+package org.apache.aries.osgi.functional;
+
+import java.util.function.Function;
+
+/**
+ * @author Carlos Sierra Andrés
+ */
+public interface Transformer<T, R> extends
+ Function<Function<R, Runnable>, Function<T, Runnable>> {
+}
Modified:
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/OSGiImpl.java
URL:
http://svn.apache.org/viewvc/aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/OSGiImpl.java?rev=1815578&r1=1815577&r2=1815578&view=diff
==============================================================================
---
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/OSGiImpl.java
(original)
+++
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/OSGiImpl.java
Fri Nov 17 15:59:59 2017
@@ -61,6 +61,13 @@ public class OSGiImpl<T> implements OSGi
}
@Override
+ public <S> OSGi<S> transformer(
+ Function<Function<S, Runnable>, Function<T, Runnable>> fun) {
+
+ return new TransformerOSGi<>(this, fun);
+ }
+
+ @Override
public OSGi<T> recover(BiFunction<T, Exception, T> onError) {
return new OSGiImpl<>((bundleContext, op) ->
_operation.run(
@@ -226,11 +233,6 @@ public class OSGiImpl<T> implements OSGi
}
@Override
- public OSGi<T> route(Consumer<Router<T>> routerConsumer) {
- return new RouteOsgiImpl<>(this, routerConsumer);
- }
-
- @Override
public <S> OSGi<S> applyTo(OSGi<Function<T, S>> fun) {
return new OSGiImpl<>(
(bundleContext, op) -> {
Added:
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/TransformerOSGi.java
URL:
http://svn.apache.org/viewvc/aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/TransformerOSGi.java?rev=1815578&view=auto
==============================================================================
---
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/TransformerOSGi.java
(added)
+++
aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/TransformerOSGi.java
Fri Nov 17 15:59:59 2017
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.aries.osgi.functional.internal;
+
+import java.util.function.Function;
+
+/**
+ * @author Carlos Sierra Andrés
+ */
+public class TransformerOSGi<T, R> extends OSGiImpl<R> {
+
+ public TransformerOSGi(
+ OSGiImpl<T> previous,
+ Function<Function<R, Runnable>, Function<T, Runnable>> fun) {
+
+ super((bundleContext, op) ->
+ previous._operation.run(bundleContext, fun.apply(op)));
+ }
+
+}
Modified:
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/ComponentTest.java
URL:
http://svn.apache.org/viewvc/aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/ComponentTest.java?rev=1815578&r1=1815577&r2=1815578&view=diff
==============================================================================
---
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/ComponentTest.java
(original)
+++
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/ComponentTest.java
Fri Nov 17 15:59:59 2017
@@ -50,8 +50,8 @@ import static org.apache.aries.osgi.func
import static org.apache.aries.osgi.functional.OSGi.just;
import static org.apache.aries.osgi.functional.OSGi.onClose;
import static org.apache.aries.osgi.functional.OSGi.register;
+import static org.apache.aries.osgi.functional.OSGi.serviceReferences;
import static org.apache.aries.osgi.functional.OSGi.services;
-import static
org.apache.aries.osgi.functional.test.HighestRankingRouter.highest;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
@@ -394,6 +394,13 @@ public class ComponentTest {
}
+ private static <T> OSGi<CachingServiceReference<T>> highest(
+ Class<T> clazz) {
+
+ return serviceReferences(clazz).transformer(
+ new HighestRankingTransformer<>());
+ }
+
private class Service {}
private class ServiceOptional {}
Modified:
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/DSLTest.java
URL:
http://svn.apache.org/viewvc/aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/DSLTest.java?rev=1815578&r1=1815577&r2=1815578&view=diff
==============================================================================
---
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/DSLTest.java
(original)
+++
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/DSLTest.java
Fri Nov 17 15:59:59 2017
@@ -39,6 +39,7 @@ import java.util.Arrays;
import java.util.Dictionary;
import java.util.HashMap;
import java.util.Hashtable;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -46,16 +47,15 @@ import java.util.concurrent.atomic.Atomi
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
+import static org.apache.aries.osgi.functional.OSGi.NOOP;
import static org.apache.aries.osgi.functional.OSGi.configuration;
import static org.apache.aries.osgi.functional.OSGi.configurations;
import static org.apache.aries.osgi.functional.OSGi.just;
-import static org.apache.aries.osgi.functional.OSGi.nothing;
import static org.apache.aries.osgi.functional.OSGi.onClose;
import static org.apache.aries.osgi.functional.OSGi.once;
import static org.apache.aries.osgi.functional.OSGi.register;
import static org.apache.aries.osgi.functional.OSGi.serviceReferences;
import static org.apache.aries.osgi.functional.OSGi.services;
-import static
org.apache.aries.osgi.functional.test.HighestRankingRouter.highest;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -799,38 +799,10 @@ public class DSLTest {
}
}
- /*@Test
- public void testRouteWithError() {
- ArrayList<Object> result = new ArrayList<>();
- ArrayList<Object> left = new ArrayList<>();
-
- OSGi<Integer> program = just(
- Arrays.asList(1, 2, 3, 4, 5, 6)
- ).recoverWith(
- (__, e) -> just(0)
- ).route(router -> {
- AtomicReference<SentEvent<Integer>> sentEvent =
- new AtomicReference<>();
-
- router.onIncoming(event -> {
- sentEvent.set(router.signalAdd(event));
- });
- router.onLeaving(__ -> sentEvent.get().terminate());
- }).
- effects(__ -> {}, left::add).
- flatMap(t -> {
- if (t % 2 != 0) {
- throw new RuntimeException();
- }
-
- return just(t);
- });
-
- try (OSGiResult run = program.run(bundleContext, result::add)) {
- assertEquals(Arrays.asList(0, 2, 0, 4, 0, 6), result);
- assertEquals(Arrays.asList(1, 3, 5), left);
- }
- }*/
+ private static <T> OSGi<CachingServiceReference<T>> highest(Class<T>
clazz) {
+ return serviceReferences(clazz).transformer(
+ new HighestRankingTransformer<>());
+ }
private class Service {}
Added:
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/HighestRankingTransformer.java
URL:
http://svn.apache.org/viewvc/aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/HighestRankingTransformer.java?rev=1815578&view=auto
==============================================================================
---
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/HighestRankingTransformer.java
(added)
+++
aries/trunk/component-dsl/itests/src/main/java/org/apache/aries/osgi/functional/test/HighestRankingTransformer.java
Fri Nov 17 15:59:59 2017
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.aries.osgi.functional.test;
+
+import org.apache.aries.osgi.functional.Transformer;
+
+import java.util.Comparator;
+import java.util.PriorityQueue;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+import static org.apache.aries.osgi.functional.OSGi.NOOP;
+
+/**
+ * @author Carlos Sierra Andrés
+ */
+public class HighestRankingTransformer<T extends Comparable<? super T>>
+ implements Transformer<T, T> {
+
+ @Override
+ public Function<T, Runnable> apply(Function<T, Runnable> publisher) {
+ PriorityQueue<T> set = new PriorityQueue<>(Comparator.reverseOrder());
+ AtomicReference<Runnable> terminator = new AtomicReference<>(NOOP);
+
+ return t -> {
+ synchronized (set) {
+ set.add(t);
+
+ if (set.peek() == t) {
+ Runnable old = terminator.get();
+
+ old.run();
+
+ terminator.set(publisher.apply(t));
+ }
+ }
+
+ return () -> {
+ synchronized (set) {
+ T old = set.peek();
+
+ set.remove(t);
+
+ T current = set.peek();
+
+ if (current != old) {
+ terminator.getAndSet(NOOP).run();
+
+ if (current != null) {
+ terminator.set(publisher.apply(current));
+ }
+ }
+ }
+ };
+ };
+ }
+}
