Merge pull request #318 from benjchristensen/async-observable

HystrixAsyncCommand and HystrixObservableCommand

hystrix-core/src/main/java/com/netflix/hystrix/HystrixCollapser.java

@@ -23,8 +23,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import rx.*;
-import rx.Observable.OnSubscribe;
+import rx.Observable;
+import rx.Scheduler;
 import rx.functions.Func1;
 import rx.schedulers.Schedulers;
 import rx.subjects.ReplaySubject;
@@ -59,7 +59,7 @@
  * @param <RequestArgumentType>
  *            The type of the request argument. If multiple arguments are needed, wrap them in another object or a Tuple.
  */
-public abstract class HystrixCollapser<BatchReturnType, ResponseType, RequestArgumentType> implements HystrixExecutable<ResponseType> {
+public abstract class HystrixCollapser<BatchReturnType, ResponseType, RequestArgumentType> implements HystrixExecutable<ResponseType>, HystrixObservable<ResponseType> {
 
     static final Logger logger = LoggerFactory.getLogger(HystrixCollapser.class);
 

hystrix-core/src/main/java/com/netflix/hystrix/HystrixCollapserProperties.java

@@ -15,7 +15,7 @@
  */
 package com.netflix.hystrix;
 
-import static com.netflix.hystrix.strategy.properties.HystrixProperty.Factory.*;
+import static com.netflix.hystrix.strategy.properties.HystrixProperty.Factory.asProperty;
 
 import com.netflix.hystrix.strategy.properties.HystrixPropertiesChainedArchaiusProperty;
 import com.netflix.hystrix.strategy.properties.HystrixPropertiesStrategy;

hystrix-core/src/main/java/com/netflix/hystrix/HystrixCommandProperties.java

@@ -15,7 +15,7 @@
  */
 package com.netflix.hystrix;
 
-import static com.netflix.hystrix.strategy.properties.HystrixProperty.Factory.*;
+import static com.netflix.hystrix.strategy.properties.HystrixProperty.Factory.asProperty;
 
 import java.util.concurrent.Future;
 

hystrix-core/src/main/java/com/netflix/hystrix/HystrixExecutable.java

@@ -29,7 +29,7 @@
  * 
  * @param <R>
  */
-public interface HystrixExecutable<R> {
+public interface HystrixExecutable<R> extends HystrixInvokable<R> {
 
     /**
      * Used for synchronous execution of command.

hystrix-core/src/main/java/com/netflix/hystrix/HystrixExecutableInfo.java

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * 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 com.netflix.hystrix;
 
 import java.util.List;

hystrix-core/src/main/java/com/netflix/hystrix/HystrixInvokable.java

@@ -0,0 +1,8 @@
+package com.netflix.hystrix;
+
+/**
+ * Marker interface for Hystrix commands that can be invoked.
+ */
+public interface HystrixInvokable<R> {
+
+}

hystrix-core/src/main/java/com/netflix/hystrix/HystrixObservable.java

@@ -0,0 +1,96 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * 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 com.netflix.hystrix;
+
+import rx.Observable;
+import rx.schedulers.Schedulers;
+
+import com.netflix.hystrix.HystrixCommandProperties.ExecutionIsolationStrategy;
+import com.netflix.hystrix.exception.HystrixBadRequestException;
+import com.netflix.hystrix.exception.HystrixRuntimeException;
+
+/**
+ * Common interface for executables that implement the Observable methods {@link #observe()} and {@link #toObservable()} so client code can treat them the same and combine in typed collections if desired.
+ * 
+ * @param <R>
+ */
+public interface HystrixObservable<R> extends HystrixInvokable<R> {
+
+    /**
+     * Used for asynchronous execution of command with a callback by subscribing to the {@link Observable}.
+     * <p>
+     * This eagerly starts execution of the command the same as {@link #queue()} and {@link #execute()}.
+     * <p>
+     * A lazy {@link Observable} can be obtained from {@link #toObservable()}.
+     * <p>
+     * <b>Callback Scheduling</b>
+     * <p>
+     * <ul>
+     * <li>When using {@link ExecutionIsolationStrategy#THREAD} this defaults to using {@link Schedulers#threadPoolForComputation()} for callbacks.</li>
+     * <li>When using {@link ExecutionIsolationStrategy#SEMAPHORE} this defaults to using {@link Schedulers#immediate()} for callbacks.</li>
+     * </ul>
+     * Use {@link HystrixCommand#toObservable(rx.Scheduler)} or {@link HystrixCollapser#toObservable(rx.Scheduler)} to schedule the callback differently.
+     * <p>
+     * See https://github.com/ReactiveX/RxJava/wiki for more information.
+     * 
+     * @return {@code Observable<R>} that executes and calls back with the result of {@link #run()} execution or a fallback from {@link #getFallback()} if the command fails for any reason.
+     * @throws HystrixRuntimeException
+     *             if a fallback does not exist
+     *             <p>
+     *             <ul>
+     *             <li>via {@code Observer#onError} if a failure occurs</li>
+     *             <li>or immediately if the command can not be queued (such as short-circuited, thread-pool/semaphore rejected)</li>
+     *             </ul>
+     * @throws HystrixBadRequestException
+     *             via {@code Observer#onError} if invalid arguments or state were used representing a user failure, not a system failure
+     * @throws IllegalStateException
+     *             if invoked more than once
+     */
+    public Observable<R> observe();
+
+    /**
+     * Used for asynchronous execution of command with a callback by subscribing to the {@link Observable}.
+     * <p>
+     * This lazily starts execution of the command only once the {@link Observable} is subscribed to.
+     * <p>
+     * An eager {@link Observable} can be obtained from {@link #observe()}
+     * <p>
+     * <b>Callback Scheduling</b>
+     * <p>
+     * <ul>
+     * <li>When using {@link ExecutionIsolationStrategy#THREAD} this defaults to using {@link Schedulers#threadPoolForComputation()} for callbacks.</li>
+     * <li>When using {@link ExecutionIsolationStrategy#SEMAPHORE} this defaults to using {@link Schedulers#immediate()} for callbacks.</li>
+     * </ul>
+     * Use {@link HystrixCommand#toObservable(rx.Scheduler)} or {@link HystrixCollapser#toObservable(rx.Scheduler)} to schedule the callback differently.
+     * <p>
+     * See https://github.com/ReactiveX/RxJava/wiki for more information.
+     * 
+     * @return {@code Observable<R>} that executes and calls back with the result of {@link #run()} execution or a fallback from {@link #getFallback()} if the command fails for any reason.
+     * @throws HystrixRuntimeException
+     *             if a fallback does not exist
+     *             <p>
+     *             <ul>
+     *             <li>via {@code Observer#onError} if a failure occurs</li>
+     *             <li>or immediately if the command can not be queued (such as short-circuited, thread-pool/semaphore rejected)</li>
+     *             </ul>
+     * @throws HystrixBadRequestException
+     *             via {@code Observer#onError} if invalid arguments or state were used representing a user failure, not a system failure
+     * @throws IllegalStateException
+     *             if invoked more than once
+     */
+    public Observable<R> toObservable();
+
+}

hystrix-core/src/main/java/com/netflix/hystrix/HystrixObservableCollapser.java

@@ -15,21 +15,29 @@
  */
 package com.netflix.hystrix;
 
-import java.util.*;
-import java.util.concurrent.*;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
-import org.slf4j.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import rx.*;
 import rx.Observable;
-import rx.functions.*;
+import rx.Scheduler;
+import rx.functions.Action0;
+import rx.functions.Func1;
 import rx.schedulers.Schedulers;
 import rx.subjects.ReplaySubject;
 
 import com.netflix.hystrix.HystrixCollapser.CollapsedRequest;
 import com.netflix.hystrix.HystrixCommandProperties.ExecutionIsolationStrategy;
-import com.netflix.hystrix.collapser.*;
-import com.netflix.hystrix.exception.HystrixRuntimeException;
+import com.netflix.hystrix.collapser.CollapserTimer;
+import com.netflix.hystrix.collapser.HystrixCollapserBridge;
+import com.netflix.hystrix.collapser.RealCollapserTimer;
+import com.netflix.hystrix.collapser.RequestCollapser;
+import com.netflix.hystrix.collapser.RequestCollapserFactory;
 import com.netflix.hystrix.strategy.HystrixPlugins;
 import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;
 import com.netflix.hystrix.strategy.properties.HystrixPropertiesStrategy;
@@ -55,7 +63,7 @@
  * @param <RequestArgumentType>
  *            The type of the request argument. If multiple arguments are needed, wrap them in another object or a Tuple.
  */
-public abstract class HystrixObservableCollapser<K, BatchReturnType, ResponseType, RequestArgumentType> implements HystrixExecutable<ResponseType> {
+public abstract class HystrixObservableCollapser<K, BatchReturnType, ResponseType, RequestArgumentType> implements HystrixObservable<ResponseType> {
 
     static final Logger logger = LoggerFactory.getLogger(HystrixObservableCollapser.class);
 
@@ -397,52 +405,6 @@ public Observable<ResponseType> toObservable(Scheduler observeOn) {
         return response;
     }
 
-    /**
-     * Used for synchronous execution.
-     * <p>
-     * If {@link Scope#REQUEST} is being used then synchronous execution will only result in collapsing if other threads are running within the same scope.
-     * 
-     * @return ResponseType
-     *         Result of {@link HystrixCommand}{@code <BatchReturnType>} execution after passing through {@link #mapResponseToRequests} to transform the {@code <BatchReturnType>} into
-     *         {@code <ResponseType>}
-     * @throws HystrixRuntimeException
-     *             if an error occurs and a fallback cannot be retrieved
-     */
-    public ResponseType execute() {
-        try {
-            return queue().get();
-        } catch (Throwable e) {
-            if (e instanceof HystrixRuntimeException) {
-                throw (HystrixRuntimeException) e;
-            }
-            // if we have an exception we know about we'll throw it directly without the threading wrapper exception
-            if (e.getCause() instanceof HystrixRuntimeException) {
-                throw (HystrixRuntimeException) e.getCause();
-            }
-            // we don't know what kind of exception this is so create a generic message and throw a new HystrixRuntimeException
-            String message = getClass().getSimpleName() + " HystrixCollapser failed while executing.";
-            logger.debug(message, e); // debug only since we're throwing the exception and someone higher will do something with it
-            //TODO should this be made a HystrixRuntimeException?
-            throw new RuntimeException(message, e);
-        }
-    }
-
-    /**
-     * Used for asynchronous execution.
-     * <p>
-     * This will queue up the command and return a Future to get the result once it completes.
-     * 
-     * @return ResponseType
-     *         Result of {@link HystrixCommand}{@code <BatchReturnType>} execution after passing through {@link #mapResponseToRequests} to transform the {@code <BatchReturnType>} into
-     *         {@code <ResponseType>}
-     * @throws HystrixRuntimeException
-     *             within an <code>ExecutionException.getCause()</code> (thrown by {@link Future#get}) if an error occurs and a fallback cannot be retrieved
-     */
-    public Future<ResponseType> queue() {
-        final Observable<ResponseType> o = toObservable();
-        return o.toBlocking().toFuture();
-    }
-
     /**
      * Key to be used for request caching.
      * <p>
@@ -550,4 +512,4 @@ public Setter andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter p
     @SuppressWarnings("rawtypes")
     private static ConcurrentHashMap<Class<? extends HystrixObservableCollapser>, String> defaultNameCache = new ConcurrentHashMap<Class<? extends HystrixObservableCollapser>, String>();
 
-}
+}