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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
package com.softwaremill.sttp.okhttp
import java.io.IOException
import java.nio.charset.Charset
import com.softwaremill.sttp._
import com.softwaremill.sttp.model._
import okhttp3.internal.http.HttpMethod
import okhttp3.{
Call,
Callback,
MediaType,
OkHttpClient,
Request => OkHttpRequest,
RequestBody => OkHttpRequestBody,
Response => OkHttpResponse
}
import okio.{BufferedSink, Okio}
import scala.collection.JavaConverters._
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.language.higherKinds
abstract class OkHttpClientHandler[R[_], S](client: OkHttpClient)
extends SttpHandler[R, S] {
private[okhttp] def convertRequest[T](request: Request[T, S]): OkHttpRequest = {
val builder = new OkHttpRequest.Builder()
.url(request.uri.toString)
val body = setBody(request.body)
builder.method(request.method.m, body.getOrElse {
if (HttpMethod.requiresRequestBody(request.method.m))
OkHttpRequestBody.create(null, "")
else null
})
//OkHttp support automatic gzip compression
request.headers
.filter(_._1.equalsIgnoreCase(AcceptEncodingHeader) == false)
.foreach {
case (name, value) => builder.addHeader(name, value)
}
builder.build()
}
private def setBody(requestBody: RequestBody[S]): Option[OkHttpRequestBody] = {
requestBody match {
case NoBody => None
case StringBody(b, encoding) =>
Some(OkHttpRequestBody.create(MediaType.parse(encoding), b))
case ByteArrayBody(b) => Some(OkHttpRequestBody.create(null, b))
case ByteBufferBody(b) => Some(OkHttpRequestBody.create(null, b.array()))
case InputStreamBody(b) =>
Some(new OkHttpRequestBody() {
override def writeTo(sink: BufferedSink): Unit =
sink.writeAll(Okio.source(b))
override def contentType(): MediaType = null
})
case PathBody(b) => Some(OkHttpRequestBody.create(null, b.toFile))
case SerializableBody(f, t) => setBody(f(t))
case StreamBody(s) => None
}
}
private[okhttp] def readResponse[T](
res: OkHttpResponse,
responseAs: ResponseAs[T, S]): R[Response[T]] = {
val body = readResponseBody(res, responseAs)
val headers = res
.headers()
.names()
.asScala
.flatMap(name => res.headers().values(name).asScala.map((name, _)))
responseMonad.map(body, Response(_: T, res.code(), headers.toList))
}
private def readResponseBody[T](res: OkHttpResponse,
responseAs: ResponseAs[T, S]): R[T] = {
responseAs match {
case IgnoreResponse => responseMonad.unit(res.body().close())
case ResponseAsString(encoding) =>
responseMonad.unit(
res.body().source().readString(Charset.forName(encoding)))
case ResponseAsByteArray => responseMonad.unit(res.body().bytes())
case MappedResponseAs(raw, g) =>
responseMonad.map(readResponseBody(res, raw), g)
case ResponseAsStream() => responseMonad.error(new IllegalStateException("Streaming isn't supported"))
}
}
}
class OkHttpSyncClientHandler private (client: OkHttpClient)
extends OkHttpClientHandler[Id, Nothing](client) {
override def send[T](r: Request[T, Nothing]): Response[T] = {
val request = convertRequest(r)
val response = client.newCall(request).execute()
readResponse(response, r.responseAs)
}
override def responseMonad: MonadError[Id] = IdMonad
}
object OkHttpSyncClientHandler {
def apply(okhttpClient: OkHttpClient = new OkHttpClient())
: SttpHandler[Id, Nothing] =
new OkHttpSyncClientHandler(okhttpClient)
}
class OkHttpFutureClientHandler private (client: OkHttpClient)(
implicit ec: ExecutionContext)
extends OkHttpClientHandler[Future, Nothing](client) {
override def send[T](r: Request[T, Nothing]): Future[Response[T]] = {
val request = convertRequest(r)
val promise = Promise[Future[Response[T]]]()
client
.newCall(request)
.enqueue(new Callback {
override def onFailure(call: Call, e: IOException): Unit =
promise.failure(e)
override def onResponse(call: Call, response: OkHttpResponse): Unit =
promise.success(readResponse(response, r.responseAs))
})
promise.future.flatten
}
override def responseMonad: MonadError[Future] = new FutureMonad
}
object OkHttpFutureClientHandler {
def apply(okhttpClient: OkHttpClient = new OkHttpClient())(
implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
: SttpHandler[Future, Nothing] =
new OkHttpFutureClientHandler(okhttpClient)
}
|
