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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
|
package com.softwaremill.sttp.okhttp
import java.io.IOException
import java.nio.charset.Charset
import com.softwaremill.sttp._
import ResponseAs.EagerResponseHandler
import okhttp3.internal.http.HttpMethod
import okhttp3.{
Call,
Callback,
Headers,
MediaType,
OkHttpClient,
MultipartBody => OkHttpMultipartBody,
Request => OkHttpRequest,
RequestBody => OkHttpRequestBody,
Response => OkHttpResponse
}
import okio.{BufferedSink, Okio}
import scala.collection.JavaConverters._
import scala.concurrent.{ExecutionContext, Future}
import scala.language.higherKinds
import scala.util.{Failure, Try}
abstract class OkHttpHandler[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 = bodyToOkHttp(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 bodyToOkHttp[T](body: RequestBody[S]): Option[OkHttpRequestBody] = {
body match {
case NoBody => None
case StringBody(b, _, _) =>
Some(OkHttpRequestBody.create(null, 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 StreamBody(s) =>
streamToRequestBody(s)
case MultipartBody(ps) =>
val b = new OkHttpMultipartBody.Builder()
.setType(OkHttpMultipartBody.FORM)
ps.foreach(addMultipart(b, _))
Some(b.build())
}
}
private def addMultipart(builder: OkHttpMultipartBody.Builder,
mp: Multipart): Unit = {
val allHeaders = mp.additionalHeaders + (ContentDispositionHeader -> mp.contentDispositionHeaderValue)
val headers = Headers.of(allHeaders.asJava)
bodyToOkHttp(mp.body).foreach(builder.addPart(headers, _))
}
private[okhttp] def readResponse[T](
res: OkHttpResponse,
responseAs: ResponseAs[T, S]): R[Response[T]] = {
val body = responseHandler(res).handle(responseAs, responseMonad)
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 responseHandler(res: OkHttpResponse) =
new EagerResponseHandler[S] {
override def handleBasic[T](bra: BasicResponseAs[T, S]): Try[T] =
bra match {
case IgnoreResponse =>
Try(res.close())
case ResponseAsString(encoding) =>
val body = Try(
res.body().source().readString(Charset.forName(encoding)))
res.close()
body
case ResponseAsByteArray =>
val body = Try(res.body().bytes())
res.close()
body
case ras @ ResponseAsStream() =>
responseBodyToStream(res).map(ras.responseIsStream)
case ResponseAsFile(file, overwrite) =>
val body = Try(
ResponseAs.saveFile(file, res.body().byteStream(), overwrite))
res.close()
body
}
}
def streamToRequestBody(stream: S): Option[OkHttpRequestBody] = None
def responseBodyToStream(res: OkHttpResponse): Try[S] =
Failure(new IllegalStateException("Streaming isn't supported"))
}
class OkHttpSyncHandler private (client: OkHttpClient)
extends OkHttpHandler[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.response)
}
override def responseMonad: MonadError[Id] = IdMonad
}
object OkHttpSyncHandler {
def apply(okhttpClient: OkHttpClient = new OkHttpClient())
: SttpHandler[Id, Nothing] =
new OkHttpSyncHandler(okhttpClient)
}
abstract class OkHttpAsyncHandler[R[_], S](client: OkHttpClient,
rm: MonadAsyncError[R])
extends OkHttpHandler[R, S](client) {
override def send[T](r: Request[T, S]): R[Response[T]] = {
val request = convertRequest(r)
rm.flatten(rm.async[R[Response[T]]] { cb =>
def success(r: R[Response[T]]) = cb(Right(r))
def error(t: Throwable) = cb(Left(t))
client
.newCall(request)
.enqueue(new Callback {
override def onFailure(call: Call, e: IOException): Unit =
error(e)
override def onResponse(call: Call, response: OkHttpResponse): Unit =
try success(readResponse(response, r.response))
catch { case e: Exception => error(e) }
})
})
}
override def responseMonad: MonadError[R] = rm
}
class OkHttpFutureHandler private (client: OkHttpClient)(
implicit ec: ExecutionContext)
extends OkHttpAsyncHandler[Future, Nothing](client, new FutureMonad) {}
object OkHttpFutureHandler {
def apply(okhttpClient: OkHttpClient = new OkHttpClient())(
implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
: SttpHandler[Future, Nothing] =
new OkHttpFutureHandler(okhttpClient)
}
|
