ObservableCommand with Multiple Values

[benjchristensen]

Should a HystrixObservableCommand be allowed to onNext more than 1 value?

I would like it to, but it has some challenges.

The execute() and queue() methods on HystrixObservableCommand expect a single value, whereas observe()/toObservable() can handle multiple.

This means the generic type T is wrong if the Observable chooses to return more than one.

A HystrixObservableCommand<T> would actually need to return like this:

• List<T> execute()
• Future<List<T>> queue()
• Observable<T> observe()

But most of the time this would not be wanted, so it would almost necessitate the generics defining it like this: HystrixObservableCommand<T, List<T>> or HystrixObservableCommand<T, T> to keep it scalar.

I'm not a fan of either of these.

So should we prohibit multi-valued responses? If so that defeats one of the benefits of Observable and the ability to stream a response back, such as using SSE.

That said, multi-valued streaming responses also make the concept of timeout a little more confusing ... is it a timeout for the start of a response, or the entire response. I think it should be the entire response (and that's what it is right now).

Right now the code assumes a single generic like HystrixObservableCommand<T> and allows multi-valued responses when using observe()/toObservable() but will blow up if execute() or queue() are used like this (shortened):

Caused by: java.util.concurrent.ExecutionException: Observable onError
    at rx.internal.operators.BlockingOperatorToFuture$2.getValue(BlockingOperatorToFuture.java:120)
    at rx.internal.operators.BlockingOperatorToFuture$2.get(BlockingOperatorToFuture.java:112)
    at com.netflix.hystrix.HystrixExecutableBase$1.performBlockingGetWithTimeout(HystrixExecutableBase.java:488)
    at com.netflix.hystrix.HystrixExecutableBase$1.get(HystrixExecutableBase.java:387)
    at com.netflix.hystrix.HystrixExecutableBase.execute(HystrixExecutableBase.java:296)
    ... 27 more
Caused by: java.lang.IllegalArgumentException: Sequence contains too many elements

I'd appreciate input on what direction to take this.

[benjchristensen]

/cc @KoltonAndrus

[benjchristensen]

Since generics alone can't solve this (due to type erasure), if we want to support multi-valued responses we likely need another type so we'd end up with:

• HystrixObservableCommand for single-value responses
• HystrixObservableMultiValuedCommand (or something like that) for multi-valued responses and it would change the execute() and queue() methods to return List<T>

[benjchristensen]

I'm going to keep HystrixObservableCommand as single-valued so move this to 1.4.x as this can be done additively.

[benjchristensen]

The decisions here affect #302

[spencergibb]

I think the separate class is the way to go. Would Collection be better to return for execute() and queue()? Might also be useful to have an extension that supports Java 8 Stream.

[benjchristensen]

Hi @spencergibb, thanks for the feedback.

Not sure about Collection since it will always be collected into an ordered List internally. The benefit of List is that someone can use index positions if they want. What value do you see in using Collection instead of List?

I'm also curious as to how Stream would benefit this? Hystrix can't depend on Java 8, but even if it could, Collection.stream() gives a Stream so why return a Stream directly?

[benjchristensen]

Asking around it seems there is agreement that we should have 2 types:

• HystrixObservableCommand<T>
  • T execute()
  • Future<T> queue()
  • Observable<T> observe()
• HystrixObservableMultiValueCommand<T>
  • List<T> execute()
  • Future<List<T>> queue()
  • Observable<T> observe()

Are these the correct names and signatures?

[spencergibb]

@benjchristensen if you are using List internally, then go with it. If you had a Collection and were creating a List, then Collection would make sense. I was just curious about Stream. Given you can get to Stream from a Collection explicit support doesn't make sense.

[benjchristensen]

@spencergibb thanks ... and what do you think about the naming of these two types? Have better ideas? I played around with "Scalar" and "Vector" but don't like those in this context, and feel it's not necessary to say "SingleValued" but need to differentiate the one with "MultiValued" support somehow.

If we didn't have the execute and queue methods we obviously wouldn't need this as Observable automatically can support either. We are artificially limiting it for the single-valued version.

[spencergibb]

I agree, you don't need to call out "SingleValued." My only other thought was "List" because it is called out in execute and queue except that breaks down for observe.

[benjchristensen]

This change isn't straight-forward. The HystrixExecutable<R> interface won't work for this new type as R is used by execute, queue and observe.

I can't change that interface as that would break everything.

This means something like HystrixObservableMultiValueCommand won't be able to implement that interface.

[benjchristensen]

Talking through this with @KoltonAndrus it is apparent that mixing the single/multi-valued paradigms is confusing and dangerous. The key confusion is what to do if a timeout occurs partway through streaming results. What does a fallback mean in that case?

Due to this, my proposal is as follows:

• change HystrixObservableCommand to HystrixAsyncCommand
• change HystrixObservableCollapser to HystrixAsyncCollapser
• leave both of these implementing HystrixExecutable with scalar (single value) responses
• change the HystrixAsyncCommand.run() method to return a callback of some form rather than Observable so there is no confusion about multi-valued responses

This will then support non-blocking, async data sources, and be fully consumable via Rx Observables, but be strictly single-valued.

We can then do a v1.5 release where we add something like HystrixStreamingCommand that adopts the true multi-value nature of Observable and separates itself from the HystrixExecutable generics and doesn't try and implement the execute and queue methods.

[mattrjacobs]

Where this stands:

• HystrixCommand<T> supports T execute(), Future<T> queue(), Observable<T> observe(), and Observable<T> toObservable(). This always represents a scalar value.
• HystrixObservableCommand<T> supports Observable<T> observe() and Observable<T> toObservable(). This represents a stream of values.

There's a new issue, scheduled for 1.5, tracking the HystrixFutureCommand/HystrixAsyncCommand