You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aries.apache.org by cs...@apache.org on 2017/08/25 11:21:38 UTC

svn commit: r1806156 - in /aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional: OSGi.java internal/DistributeOSGi.java

Author: csierra
Date: Fri Aug 25 11:21:38 2017
New Revision: 1806156

URL: http://svn.apache.org/viewvc?rev=1806156&view=rev
Log:
Looks like `all` should combine the results

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/DistributeOSGi.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=1806156&r1=1806155&r2=1806156&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 Aug 25 11:21:38 2017
@@ -163,8 +163,9 @@ public interface OSGi<T> extends OSGiRun
 		return new ServiceReferenceOSGi<>(filterString, clazz);
 	}
 
-	static OSGi<Void> all(OSGi<?> ... programs) {
-		return new DistributeOSGi(programs);
+	@SafeVarargs
+	static <T> OSGi<T> all(OSGi<T> ... programs) {
+		return new DistributeOSGi<>(programs);
 	}
 
 	OSGi<T> filter(Predicate<T> predicate);

Modified: aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/DistributeOSGi.java
URL: http://svn.apache.org/viewvc/aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/DistributeOSGi.java?rev=1806156&r1=1806155&r2=1806156&view=diff
==============================================================================
--- aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/DistributeOSGi.java (original)
+++ aries/trunk/component-dsl/component-dsl/src/main/java/org/apache/aries/osgi/functional/internal/DistributeOSGi.java Fri Aug 25 11:21:38 2017
@@ -29,33 +29,41 @@ import java.util.stream.Collectors;
 /**
  * @author Carlos Sierra Andrés
  */
-public class DistributeOSGi extends OSGiImpl<Void> {
+public class DistributeOSGi<T> extends OSGiImpl<T> {
 
-    public DistributeOSGi(OSGi<?>... programs) {
+    @SafeVarargs
+    public DistributeOSGi(OSGi<T>... programs) {
         super(bundleContext -> {
-            Pipe<Tuple<Void>, Tuple<Void>> added = Pipe.create();
+            Pipe<Tuple<T>, Tuple<T>> added = Pipe.create();
 
-            Consumer<Tuple<Void>> addedSource = added.getSource();
+            Consumer<Tuple<T>> addedSource = added.getSource();
 
-            List<OSGiResult<?>> results = new ArrayList<>();
+            List<OSGiResult<T>> results = new ArrayList<>();
 
-            Pipe<Tuple<Void>, Tuple<Void>> removed = Pipe.create();
+            Pipe<Tuple<T>, Tuple<T>> removed = Pipe.create();
 
-            Consumer<Tuple<Void>> removedSource = removed.getSource();
+            Consumer<Tuple<T>> removedSource = removed.getSource();
 
             return new OSGiResultImpl<>(
                 added, removed,
                 () -> {
                     results.addAll(
                         Arrays.stream(programs).
-                            map(o -> o.run(bundleContext)).
-                            collect(Collectors.toList()));
+                            map(o -> {
+                                OSGiResultImpl<T> osGiResult =
+                                    ((OSGiImpl<T>) o)._operation.run(
+                                        bundleContext);
+
+                                osGiResult.added.map(t -> {addedSource.accept(t); return null;});
+                                osGiResult.removed.map(t -> {removedSource.accept(t); return null;});
+
+                                osGiResult.start.run();
 
-                    addedSource.accept(Tuple.create(null));
+                                return osGiResult;
+                            }).
+                            collect(Collectors.toList()));
                 },
                 () -> {
-                    removedSource.accept(Tuple.create(null));
-
                     for (OSGiResult<?> result : results) {
                         try {
                             result.close();