1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
package com.softwaremill.sttp.hystrix
import com.netflix.hystrix.HystrixObservableCommand.Setter
import com.netflix.hystrix.{HystrixCommand, HystrixCommandGroupKey, HystrixCommandProperties, HystrixObservableCommand}
import com.softwaremill.sttp.{MonadAsyncError, MonadError, Request, Response, SttpBackend}
import rx.{Observable, Subscriber}
class HystrixBackend[R[_], S] private (delegate: SttpBackend[R, S])(groupKey: String,
setter: HystrixCommandProperties.Setter)
extends SttpBackend[R, S] {
class AsyncSttpCMD[T](request: Request[T, S], delegate: SttpBackend[R, S])
extends HystrixObservableCommand[Response[T]](
Setter
.withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))
.andCommandPropertiesDefaults(setter)
) {
override def construct(): Observable[Response[T]] = Observable.create(
(t: Subscriber[_ >: Response[T]]) => {
val x = responseMonad.flatMap(delegate.send(request)) { response =>
t.onNext(response)
t.onCompleted()
responseMonad.unit(response)
}
responseMonad.handleError(x) {
case e: Throwable =>
t.onError(e)
responseMonad.error(e)
}
}
)
}
class SyncSttpCMD[T](request: Request[T, S], delegate: SttpBackend[R, S])
extends HystrixCommand[R[Response[T]]](
HystrixCommand.Setter
.withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))
.andCommandPropertiesDefaults(setter)
) {
override def run(): R[Response[T]] = delegate.send(request)
}
override def send[T](request: Request[T, S]): R[Response[T]] = {
responseMonad match {
case mae: MonadAsyncError[R] =>
mae.async { cb =>
new AsyncSttpCMD(request, delegate)
.observe()
.subscribe(new Subscriber[Response[T]]() {
def onNext(item: Response[T]): Unit = {
cb(Right(item))
}
def onError(error: Throwable): Unit = cb(Left(error))
def onCompleted(): Unit = {}
})
}
case _ => new SyncSttpCMD(request, delegate).execute()
}
}
override def close(): Unit = delegate.close()
/**
* The monad in which the responses are wrapped. Allows writing wrapper
* backends, which map/flatMap over the return value of [[send]].
*/
override def responseMonad: MonadError[R] = delegate.responseMonad
}
object HystrixBackend {
def apply[R[_], S](delegate: SttpBackend[R, S])(groupKey: String,
setter: HystrixCommandProperties.Setter =
HystrixCommandProperties.Setter()) =
new HystrixBackend(delegate)(groupKey, setter)
}
|
