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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
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,
propertySetter: HystrixCommandProperties.Setter)
extends SttpBackend[R, S] {
import HystrixBackend._
class AsyncSttpCMD[T](request: Request[T, S], delegate: SttpBackend[R, S])
extends HystrixObservableCommand[Response[T]](
request.tag(HystrixAsyncConfiguration).map(_.asInstanceOf[HystrixObservableCommand.Setter]) match {
case None =>
Setter
.withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))
.andCommandPropertiesDefaults(propertySetter)
case Some(s) => s
}
) {
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]]](
request.tag(HystrixSyncConfiguration).map(_.asInstanceOf[HystrixCommand.Setter]) match {
case None =>
HystrixCommand.Setter
.withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))
.andCommandPropertiesDefaults(propertySetter)
case Some(s) => s
}
) {
private var fallback: R[Response[T]] = responseMonad.unit(null)
override def run(): R[Response[T]] = {
val rr: R[Response[T]] = responseMonad.map(delegate.send(request)) {
case errResponse @ Response(Left(error), _, _, _, _) =>
fallback = responseMonad.unit(errResponse)
throw new RuntimeException(error)
case r: Response[T] => r
}
responseMonad.handleError(rr) {
case t: Throwable =>
fallback = responseMonad.error(t)
throw new RuntimeException(t)
}
}
override def getFallback: R[Response[T]] = fallback
}
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,
propertySetter: HystrixCommandProperties.Setter =
HystrixCommandProperties.Setter()) =
new HystrixBackend(delegate)(groupKey, propertySetter)
private val HystrixAsyncConfiguration = "HystrixAsyncConfiguration"
private val HystrixSyncConfiguration = "HystrixSyncConfiguration"
implicit class HystrixRichRequest[T, S](request: Request[T, S]) {
def configureAsyncHystrix(setter: HystrixObservableCommand.Setter): Request[T, S] =
request.tag(HystrixAsyncConfiguration, setter)
def configureSyncHystrix(setter: HystrixCommand.Setter): Request[T, S] =
request.tag(HystrixSyncConfiguration, setter)
}
}
|
