2 * Copyright 2011-2012 GRNET S.A. All rights reserved.
4 * Redistribution and use in source and binary forms, with or
5 * without modification, are permitted provided that the following
8 * 1. Redistributions of source code must retain the above
9 * copyright notice, this list of conditions and the following
12 * 2. Redistributions in binary form must reproduce the above
13 * copyright notice, this list of conditions and the following
14 * disclaimer in the documentation and/or other materials
15 * provided with the distribution.
17 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
18 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
19 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
20 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
21 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
22 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
23 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
24 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
25 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
26 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
27 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
28 * POSSIBILITY OF SUCH DAMAGE.
30 * The views and conclusions contained in the software and
31 * documentation are those of the authors and should not be
32 * interpreted as representing official policies, either expressed
33 * or implied, of GRNET S.A.
36 package gr.grnet.aquarium.service
38 import gr.grnet.aquarium.util._
39 import gr.grnet.aquarium.{ResourceLocator, Aquarium, Configurable, AquariumAwareSkeleton}
40 import com.ckkloverdos.props.Props
41 import com.twitter.finagle.{Service, SimpleFilter}
42 import org.jboss.netty.handler.codec.http.{HttpResponseStatus ⇒ THttpResponseStatus, DefaultHttpResponse ⇒ TDefaultHttpResponse, HttpResponse ⇒ THttpResponse, HttpRequest ⇒ THttpRequest}
43 import com.twitter.util.{Future ⇒ TFuture, FuturePool ⇒ TFuturePool, Promise ⇒ TPromise, Return ⇒ TReturn, Throw ⇒ TThrow, Duration}
44 import com.twitter.finagle.builder.ServerBuilder
45 import com.twitter.finagle.http.Http
46 import org.jboss.netty.handler.codec.http.HttpResponseStatus._
47 import org.jboss.netty.handler.codec.http.HttpVersion._
48 import org.jboss.netty.buffer.ChannelBuffers._
49 import org.jboss.netty.util.CharsetUtil._
50 import java.net.InetSocketAddress
51 import java.util.concurrent.{Executors, TimeUnit}
52 import gr.grnet.aquarium.util.date.TimeHelpers
53 import org.joda.time.format.ISODateTimeFormat
54 import gr.grnet.aquarium.actor.message.{GetUserStateRequest, GetUserBalanceRequest, RouterResponseMessage, RouterRequestMessage}
55 import gr.grnet.aquarium.actor.RouterRole
56 import com.ckkloverdos.resource.StreamResource
57 import com.ckkloverdos.maybe.{Just, Failed}
58 import gr.grnet.aquarium.event.model.ExternalEventModel
62 * @author Christos KK Loverdos <loverdos@gmail.com>
65 class FinagleRESTService extends Lifecycle with AquariumAwareSkeleton with Configurable with Loggable {
66 final val TEXT_PLAIN = "text/plain"
67 final val APPLICATION_JSON = "application/json"
69 @volatile private[this] var _port: Int = 8080
70 @volatile private[this] var _shutdownTimeoutMillis: Long = 2000
71 @volatile private[this] var _threadPoolSize: Int = 4
72 @volatile private[this] var _threadPool: TFuturePool = _
74 def propertyPrefix = Some(RESTService.Prefix)
77 * Configure this instance with the provided properties.
79 * If `propertyPrefix` is defined, then `props` contains only keys that start with the given prefix.
81 def configure(props: Props) {
82 this._port = props.getIntEx(Aquarium.EnvKeys.restPort.name)
83 this._shutdownTimeoutMillis = props.getLongEx(Aquarium.EnvKeys.restShutdownTimeoutMillis.name)
85 this._threadPool = TFuturePool(Executors.newFixedThreadPool(this._threadPoolSize))
87 logger.debug("HTTP port is %s".format(this._port))
90 def stringResponse(status: THttpResponseStatus, body: String, contentType: String) = {
91 val response = new TDefaultHttpResponse(HTTP_1_1, status)
92 response.setContent(copiedBuffer(body, UTF_8))
93 response.setHeader("Content-type", "%s;charset=utf-8".format(contentType))
95 TFuture.value(response)
98 def stringResponseOK(body: String, contentType: String): TFuture[THttpResponse] = {
99 stringResponse(OK, body, contentType)
102 def statusResponse(status: THttpResponseStatus): TFuture[THttpResponse] = {
103 stringResponse(status, status.getReasonPhrase, TEXT_PLAIN)
106 def resourceInfoResponse(resource: StreamResource, contentType: String): TFuture[THttpResponse] = {
107 val fmt = (body: String) ⇒ "%s\n\n%s".format(resource.url, body)
109 resource.stringContent.toMaybeEither match {
111 stringResponseOK(fmt(body), contentType)
118 def eventInfoResponse[E <: ExternalEventModel](
120 getter: String ⇒ Option[E]
121 ): TFuture[THttpResponse] = {
122 getter(eventID) match {
124 stringResponseOK(event.toJsonString, APPLICATION_JSON)
127 statusResponse(NOT_FOUND)
131 final case class ExceptionHandler() extends SimpleFilter[THttpRequest, THttpResponse] {
132 def apply(request: THttpRequest, service: Service[THttpRequest, THttpResponse]): TFuture[THttpResponse] = {
133 service(request) handle {
135 logger.error("While serving %s".format(request), error)
136 val statusCode = error match {
137 case _: IllegalArgumentException ⇒
140 INTERNAL_SERVER_ERROR
143 val errorResponse = new TDefaultHttpResponse(HTTP_1_1, statusCode)
144 errorResponse.setContent(copiedBuffer(error.getStackTraceString, UTF_8))
151 final case class AdminChecker() extends SimpleFilter[THttpRequest, THttpResponse] {
152 def apply(request: THttpRequest, service: Service[THttpRequest, THttpResponse]): TFuture[THttpResponse] = {
153 if(request.getUri.startsWith(RESTPaths.AdminPrefix)) {
154 val headerValue = request.getHeader(Aquarium.HTTP.RESTAdminHeaderName)
155 aquarium.adminCookie match {
156 case Some(`headerValue`) ⇒
160 statusResponse(FORBIDDEN)
168 final case class MainService() extends Service[THttpRequest, THttpResponse] {
169 final class ActorRouterService extends Service[RouterRequestMessage, RouterResponseMessage[_]] {
170 def apply(message: RouterRequestMessage): TFuture[RouterResponseMessage[_]] = {
171 // We want to asynchronously route the message via akka and get the whole computation as a
173 val actorProvider = aquarium.actorProvider
174 val router = actorProvider.actorForRole(RouterRole)
175 val promise = new TPromise[RouterResponseMessage[_]]()
177 val actualWork = router.ask(message)
179 actualWork onComplete { akkaFuture ⇒
180 akkaFuture.value match {
181 case Some(eitherValue) ⇒
183 case Left(throwable) ⇒
184 promise.setException(throwable)
187 promise.setValue(value.asInstanceOf[RouterResponseMessage[_]])
191 promise.setException(new Exception("Got no response for %s".format(message)))
199 final val actorRouterService = new ActorRouterService
201 def callRouter(requestMessage: RouterRequestMessage): TFuture[THttpResponse] = {
202 actorRouterService(requestMessage).transform { tryResponse ⇒
204 case TReturn(responseMessage: RouterResponseMessage[_]) ⇒
205 val statusCode = responseMessage.suggestedHTTPStatus
206 val status = THttpResponseStatus.valueOf(statusCode)
208 responseMessage.response match {
209 case Left(errorMessage) ⇒
210 logger.error("Error %s '%s' serving %s. Internal response: %s".format(
216 stringResponse(status, errorMessage, TEXT_PLAIN)
219 stringResponse(status, responseMessage.toJsonString, APPLICATION_JSON)
222 case TThrow(throwable) ⇒
223 val status = INTERNAL_SERVER_ERROR
224 logger.error("Error %s serving %s: %s".format(
225 status.getReasonPhrase,
227 gr.grnet.aquarium.util.shortInfoOf(throwable)
230 statusResponse(status)
235 def apply(request: THttpRequest): TFuture[THttpResponse] = {
236 val millis = TimeHelpers.nowMillis()
237 val uri = request.getUri
238 val method = request.getMethod
239 logger.debug("%s %s".format(method, uri))
242 case RESTPaths.PingPath() ⇒
243 val now = TimeHelpers.nowMillis()
244 val nowFormatted = ISODateTimeFormat.dateTime().print(now)
245 stringResponseOK("PONG\n%s\n%s".format(now, nowFormatted), TEXT_PLAIN)
247 // case RESTPaths.ResourcesPath() ⇒
248 // stringResponseOK("%s\n%s\n%s\n" .format(
249 // ResourceLocator.ResourceNames.AQUARIUM_PROPERTIES,
250 // ResourceLocator.ResourceNames.LOGBACK_XML,
251 // ResourceLocator.ResourceNames.POLICY_YAML),
254 case RESTPaths.UserBalancePath(userID) ⇒
255 // /user/(.+)/balance/?
256 callRouter(GetUserBalanceRequest(userID, millis))
258 case RESTPaths.UserStatePath(userId) ⇒
259 // /user/(.+)/state/?
260 callRouter(GetUserStateRequest(userId, millis))
262 case RESTPaths.ResourcesAquariumPropertiesPath() ⇒
263 resourceInfoResponse(ResourceLocator.Resources.AquariumPropertiesResource, TEXT_PLAIN)
265 case RESTPaths.ResourcesLogbackXMLPath() ⇒
266 resourceInfoResponse(ResourceLocator.Resources.LogbackXMLResource, TEXT_PLAIN)
268 case RESTPaths.ResourcesPolicyYAMLPath() ⇒
269 resourceInfoResponse(ResourceLocator.Resources.PolicyYAMLResource, TEXT_PLAIN)
271 case RESTPaths.ResourceEventPath(id) ⇒
272 eventInfoResponse(id, aquarium.resourceEventStore.findResourceEventByID)
274 case RESTPaths.IMEventPath(id) ⇒
275 eventInfoResponse(id, aquarium.imEventStore.findIMEventByID)
278 statusResponse(NOT_FOUND)
283 val service = ExceptionHandler() andThen AdminChecker() andThen MainService()
284 lazy val server = ServerBuilder().
286 bindTo(new InetSocketAddress(this._port)).
290 def start(): Unit = {
291 logger.info("Starting HTTP on port %s".format(this._port))
292 // Just for the side effect
293 assert(server ne null)
297 logger.info("Stopping HTTP on port %s, waiting for at most %s ms".format(this._port, this._shutdownTimeoutMillis))
298 server.close(Duration(this._shutdownTimeoutMillis, TimeUnit.MILLISECONDS))