After 5 months studying theories deeper & deeper on my free-time and preparing 3 talks for scala.io & ping-conf with my friend Julien Tournay aka @skaalf, I’m back blogging and I’ve got a few more ideas of articles to come…
If you’re interested in those talks, you can find pingconf videos here:
This new API is just the logical extension of play2/Scala Json API (that I’ve been working & promoting those 2 last years) pushing its principles far further by allowing validation on any data types.
This new API is a real step further as it will progressively propose a common API for all validations in Play2/Scala (Form/Json/XML/…). It proposes an even more robust design relying on very strong theoretical ground making it very reliable & typesafe.
Julien has written his article presenting the new API basics and he also found time to write great documentation for this new validation API. I must confess Json API doc was quite messy but I’ve never found freetime (and courage) to do better. So I’m not going to spend time on basic features of this new API and I’m going to target advanced features to open your minds about the power of this new API.
Let’s have fun with this new APi & Shapeless, this fantastic tool for higher-rank polymorphism & type-safety!
Warm-up with Higher-kind Zipping of Rules
A really cool & new feature of Play2.3 generic validation API is its ability to compose validation Rules in chains like:
Please note the new From[JsValue]: if it were Xml, it would be From[Xml], genericity requires some more info.
Ok that’s not too hard but sometimes you would like to use first the macro and after those primary type validations, you want to refine with custom validations. Something like:
12
Rule.gen[JsValue, FooBar]+?+?+((notEmpty:Rule[String, String])+:(min(5):Rule[Int, Int])+:(min(10L):Rule[Long,Long]))// +?+?+ is a non-existing operator meaning "compose"
As you may know, you can’t do use this +: from Scala Sequence[T] as this list of Rules is typed heterogenously and Rule[I, O] is invariant.
So we are going to use Shapeless heterogenous Hlist for that:
Finally, let’s wire all together in a Play action:
12345678910111213141516171819202122232425
defhzipper=Action(parse.json){request=>ruleZip.validate(request.body)map{foobar=>Ok(foobar.toString)}recoverTotal{errors=>BadRequest(errors.toString)}}// OK case{"foo":"toto","bar":5,"foo2":5}=>toto::5::8::HNil// KO case{"foo":"","bar":2,"foo2":12}=>Failure(List(([0],List(ValidationError(error.max,WrappedArray(10)))),([1],List(ValidationError(error.min,WrappedArray(5)))),([2],List(ValidationError(error.required,WrappedArray())))))
As you can see, the problem in this approach is that we lose the path of Json.
Anyway, this can give you a few ideas!
Now let’s do something really useful…
Higher-kind Fold of Rules to break the 22 limits
As in Play2.1 Json API, the new validation API provides an applicative builder which allows the following:
But, in Play2.1 Json API and also in new validation API, all functional combinators are limited by the famous Scala 22 limits.
In Scala, you CAN’T write :
a case-class with >22 fields
a Tuple23
So you can’t do Rule[JsValue, A] ~ Rule[JsValue, B] ~ ... more than 22 times.
Nevertheless, sometimes you receive huge JSON with much more than 22 fields in it. Then you have to build more complex models like case-classes embedding case-classes… Shameful, isn’t it…
Let’s be shameless with Shapeless HList which enables to have unlimited heterogenously typed lists!
This looks like a higher-kind fold so let’s call that HFOLD.
We can build this hfold using Shapeless polymorphic functions & RighFolder.
In a next article, I may write about coding such shapeless feature. Meanwhile, you’ll have to discover the code on Github as it’s a bit hairy but very interesting ;)
A websocket is a persistent bi-directional channel of communication (in/out) and is created with:
an Iteratee[A, _] to manage all frames received by the websocket endpoint
an Enumerator[A] to send messages through the websocket
an implicit FrameFormatter[A] to parse frame content to type A (Play provides default FrameFormatter for String or JsValue)
Here is how you traditionally create a websocket endpoint in Play:
12345678910
objectMyControllerextendsController{defconnect=Websocket.async[JsValue]{rh=>// the iteratee to manage received messagesvaliteratee=Iteratee.foreach[JsValue](js=>...)// the enumerator to be able to send messagesvalenumerator=// generally a PushEnumerator(iteratee,enumerator)}}
Generally, the Enumerator[A] is created using Concurrent.broadcast[A] and Concurrent.unicast[A] which are very powerful tools but not so easy to understand exactly (the edge-cases of connection close, errors are always tricky).
You often want to:
manage multiple client connections at the same time
parse messages received from websockets,
do something with the message payload
send messages to a given client
broadcast messages to all connected members
create bots to be able to simulate fake connected members
etc…
To do that in Play non-blocking/async architecture, you often end developing an Actor topology managing all events/messages on top of the previous Iteratee/Enumerator.
The Iteratee/Enumerator is quite generic but always not so easy to write.
The actor topology is quite generic because there are administration messages that are almost always the same:
Connection/Forbidden/Disconnection
Broadcast/Send
Actor Room is a helper managing all of this for you.
So you can just focus on message management using actors and nothing else. It provides all default behaviors and all behaviors can be overriden if needed. It exposes only actors and nothing else.
The code is based on the chatroom sample (and a cool sample by Julien Tournay) from Play Framework pushed far further and in a more generic way.
What is Actor Room?
An actor room manages a group of connected members which are supervised by a supervisor
Member = 2 actors (receiver/sender)
Each member is represented by 2 actors (1 receiver & 1 sender):
You MUST create at least a Receiver Actor because it’s your job to manage your own message format
The Sender Actor has a default implementation but you can override it.
Supervisor = 1 actor
All actors are managed by 1 supervisor which have two roles:
// default constructorvalroom=Room()// constructor with custom supervisor// custom supervisor are described latervalroom=Room(Props(classOf[CustomSupervisor]))
The room creates the Supervisor actor for you and delegates the creation of receiver/sender actors to it.
If you want to broadcast a message or target a precise member, you should use the supervisor.
/** Message received and parsed to type A * @param from the ID of the sender * @param payload the content of the message */caseclassReceived[A](from:String,payload:A)extendsMessage
If your websocket frames contain Json, then it should be Received[JsValue].
You just have to create a simple actor:
12345678910111213141516
// Create an actor to receive messages from websocketclassReceiverextendsActor{defreceive={// Received(fromId, js) is the only Message to manage in receivercaseReceived(from,js:JsValue)=>(js\"msg").asOpt[String]match{caseNone=>play.Logger.error("couldn't msg in websocket event")caseSome(s)=>play.Logger.info(s"received $s")// broadcast message to all connected memberscontext.parent!Broadcast(from,Json.obj("msg"->s))}}}
Please note the Receiver Actor is supervised by the Supervisor actor. So, within the Receiver Actor, context.parent is the Supervisor and you can use it to send/broadcast message as following:
123456789
context.parent!Send(fromId,toId,mymessage)context.parent!Broadcast(fromId,mymessage)// The 2 messages/** Sends a message from a member to another member */caseclassSend[A](from:String,to:String,payload:A)extendsMessage/** Broadcasts a message from a member */caseclassBroadcast[A](from:String,payload:A)extendsMessage
Create your Json websocket endpoint
Please note that each member is identified by a string that you define yourself.
import org.mandubian.actorroom._
12345678910111213141516171819
classReceiverextendsActor{defreceive={...}}objectApplicationextendsController{valroom=Room()/** websocket requires : * - the type of the Receiver actor * - the type of the payload */defconnect(id:String)=room.websocket[Receiver, JsValue](id)// ordefconnect(id:String)=room.websocket[JsValue](id,Props[Receiver])}
importakka.actor._importplay.api._importplay.api.mvc._importplay.api.libs.json._// Implicitsimportplay.api.Play.currentimportplay.api.libs.concurrent.Execution.Implicits._importorg.mandubian.actorroom._classReceiverextendsActor{defreceive={caseReceived(from,js:JsValue)=>(js\"msg").asOpt[String]match{caseNone=>play.Logger.error("couldn't msg in websocket event")caseSome(s)=>play.Logger.info(s"received $s")context.parent!Broadcast(from,Json.obj("msg"->s))}}}objectApplicationextendsController{valroom=Room()defwebsocket(id:String)=room.websocket[Receiver, JsValue](id)}
Extend default behaviors
Override the administration message format
AdminMsgFormatter typeclass is used by ActorRoom to format administration messages (Connected, Disconnected and Error) by default.
AdminMsgFormatter[JsValue] and AdminMsgFormatter[String] are provided by default.
You can override the format as following:
123456789
// put this implicit in the same scope where you create your websocket endpointimplicitvalmsgFormatter=newAdminMsgFormatter[JsValue]{defconnected(id:String)=Json.obj("kind"->"connected","id"->id)defdisconnected(id:String)=Json.obj("kind"->"disconnected","id"->id)deferror(id:String,msg:String)=Json.obj("kind"->"error","id"->id,"msg"->msg)}// then this msgFormatter will be used for all administration messages defwebsocket(id:String)=room.websocket[Receiver, JsValue](id)
Override the Sender Actor
You just have to create a new actor as following:
12345678910111213141516
classMyCustomSenderextendsActor{defreceive={cases:Send[JsValue]=>// message send from a member to another onecaseb:Broadcast[JsValue]=>// message broadcast by a membercaseConnected(id)=>// member "id" has connectedcaseDisconnected(id)=>// member "id" has disconnectedcaseInit(id,receiverActor)=>// Message sent when sender actor is initialized by ActorRoom}}
// public sender messages/** Sender actor is initialized by Supervisor */caseclassInit(id:String,receiverActor:ActorRef)/** Sends a message from a member to another member */caseclassSend[A](from:String,to:String,payload:A)extendsMessage/** Broadcasts a message from a member */caseclassBroadcast[A](from:String,payload:A)extendsMessage/** member with ID has connected */caseclassConnected(id:String)extendsMessage/** member with ID has disconnected */caseclassDisconnected(id:String)extendsMessage
Override the Supervisor Actor
Please note Supervisor is an actor which manages a internal state containing all members:
1
varmembers=Map.empty[String, Member]
You can override the default Supervisor as following:
1234567891011121314151617
classCustomSupervisorextendsSupervisor{defcustomBroadcast:Receive={caseBroadcast(from,js:JsObject)=>// adds members to all messagesvalids=Json.obj("members"->members.map(_._1))members.foreach{case(id,member)=>member.sender!Broadcast(from,js++ids)case_=>()}}overridedefreceive=customBroadcastorElsesuper.receive}
Create a bot to simulate member
A bot is a fake member that you can use to communicate with other members. It’s identified by an ID as any member.
Then with returned Member, you can simulate messages:
123456
valroom=Room()valbot=room.bot[JsValue]("robot")// simulate a received messagebot.receiver!Received(bod.id,Json.obj("foo"->"bar"))
Naturally, you can override the Bot Sender Actor
123456789101112
/** The default actor sender for Bots */classBotSenderextendsActor{defreceive={cases=>play.Logger.info(s"Bot should have sent ${s}")}}valbot=room.bot[JsValue]("robot",Props[BotSender])
So what else???
Everything you can override and everything that I didn’t implement yet…
The code for all autosources & sample apps can be found on Github here
The aim of this article is to show how scalaz-stream could be plugged on existing Play Iteratee/Enumerator and used in your web projects. I also wanted to evaluate in depth the power of scalaz-streamProcesses by trying to write a recursive streaming action: I mean a web endpoint streaming data and re-injecting its own streamed data in itself.
If you want to see now how scalaz-stream is used with Play, go to this paragraph directly.
Why Scalaz-Stream when you have Play Iteratees?
Play Iteratees are powerful & cool but…
I’m a fan of everything dealing with data streaming and realtime management in backends. I’ve worked a lot on Play Framework and naturally I’ve been using the cornerstone behind Play’s reactive nature: Play Iteratees.
Iteratees (with its counterparts, Enumerators and Enumeratees) are great to manipulate/transform linear streams of data chunks in a very reactive (non-blocking & asynchronous) and purely functional way:
Enumerators identifies the data producer that can generate finite/infinite/procedural data streams.
Iteratee is simply a data folder built as a state machine based on 3 states (Continue, Done, Error) which consumes data from Enumerator to compute a final result.
Enumeratee is a kind of transducer able to adapt an Enumerator producing some type of data to an Iteratee that expects other type of data. Enumeratee can be used as both a pipe transformer and adapter.
Iteratee is really powerful but I must say I’ve always found them quite picky to use, practically speaking. In Play, they are used in their best use-case and they were created for that exactly. I’ve been using Iteratees for more than one year now but I still don’t feel fluent with them. Each time I use them, I must spend some time to know how I could write what I need. It’s not because they are purely functional (piping an Enumerator into an Enumeratee into an Iteratee is quite trivial) but there is something that my brain doesn’t want to catch.
If you want more details about my experience with Iteratees, go to this paragraph
That’s why I wanted to work with other functional streaming tools to see if they suffer the same kind of usability toughness or can bring something more natural to me. There are lots of other competitors on the field such as pipes, conduits and machines. As I don’t have physical time to study all of them in depth, I’ve chosen the one that appealed me the most i.e. Machines.
I’m not yet a Haskell coder even if I can mumble it so I preferred to evaluate the concept with scalaz-stream, a Scala implementation trying to bring machines to normal coders focusing on the aspect of IO streaming.
Scratching the concepts of Machine / Process ?
I’m not going to judge if Machines are better or not than Iteratees, this is not my aim. I’m just experimenting the concept in an objective way.
I won’t explain the concept of Machines in depth because it’s huge and I don’t think I have the theoretical background to do it right now. So, let’s focus on very basic ideas at first:
Machine is a very generic concept that represents a data processing mechanism with potential multiple inputs, an output and monadic effects (typically Future input chunks, side-effects while transforming, delayed output…)
To simplify, let say a machine is a bit like a mechano that you construct by plugging together other more generic machines (such as source, transducer, sink, tee, wye) as simply as pipes.
Building a machine also means planning all the steps you will go through when managing streamed data but it doesn’t do anything until you run it (no side-effect, no resource consumption). You can re-run a machine as many times as you want.
A machine is a state machine (Emit/Await/Halt) as Iteratee but it manages error in a more explicit way IMHO (fallback/error)
In scalaz-stream, you don’t manipulate machines which are too abstract for real-life use-cases but you manipulate simpler concepts:
Process[M, O] is a restricted machine outputting a stream of O. It can be a source if the monadic effect gets input from I/O or generates procedural data, or a sink if you don’t care about the output. Please note that it doesn’t infer the type of potential input at all.
Wye[L, R, O] is a machine that takes 2 inputs (left L / right R) and outputs chunks of type O (you can read from left or right or wait for both before ouputting)
Tee[L, R, O] is a Wye that can only read alternatively from left or from right but not from both at the same time.
Process1[I, O] can be seen as a transducer which accepts inputs of type I and outputs chunks of type O (a bit like Enumeratee)
Channel[M, I, O] is an effectul channel that accepts input of type I and use it in a monadic effect M to produce potential O
What I find attractive in Machines?
Machines is producer/consumer/transducer in the same place and Machines can consume/fold as Iteratee, transform as Enumeratee and emit as Enumerator at the same time and it opens lots of possibilities (even if 3 concepts in one could make it more complicated too).
I feel like playing with legos as you plug machines on machines and this is quite funny actually.
Machines manages monadic effects in its design and doesn’t infer the type of effect so you can use it with I/O, Future and whatever you can imagine that is monadic…
Machines provide out-of-the-box Tee/Wye to compose streams, interleave, zip them as you want without writing crazy code.
The early code samples I’ve seen were quite easy to read (even the implementation is not so complex). Have a look at the StartHere sample provided by scalaz-stream:
But don’t think everything is so simple, machines is a complex concept with lots of theory behind it which is quite abstract. what I find very interesting is that it’s possible to vulgarize this very abstract concept with simpler concepts such as Process, Source, Sink, Tee, Wye… that you can catch quite easily as these are concepts you already manipulated when you were playing in your bathtub when you were child (or even now).
After these considerations, I wanted to experiment scalaz-stream with Play streaming capabilities in order to see how it behaves in a context I know.
Here is what I decided to study:
Stream data out of a controller action using a scalaz-stream Process
Call an AsyncWebService & consume the response as a stream of Array[Byte] using a scalaz-stream Process
Here is existing Play API :
Action provides Ok.stream(Enumerator)
WS call consuming response as a stream of data WS.get(r: ResponseHeader => Iteratee)
As you can see, these API depends on Iteratee/Enumerator. As I didn’t want to hack Play too much as a beginning, I decided to try & plug scalaz-stream on Play Iteratee (if possible).
Building Enumerator[O] from Process[Task, O]
The idea is to take a scalaz-stream Source[O] (Process[M,O]) and wrap it into an Enumerator[O] so that it can be used in Play controller actions.
An Enumerator is a data producer which can generate those data using monadic Future effects (Play Iteratee is tightly linked to Future).
Process[Task, O] is a machine outputting a stream of O so it’s logically the right candidate to be adapted with a Enumerator[O]. Let’s remind’ Task is just a scalaz Future[Either[Throwable,A]] with a few helpers and it’s used in scalaz-stream.
So I’ve implemented (at least tried) an Enumerator[O] that accepts a Process[Task, O]:
123456
defenumerator[O](p:Process[Task, O])(implicitctx:ExecutionContext)=newEnumerator[O]{...// look the code in github project...}
The implementation just synchronizes the states of the Iteratee[O, A] consuming the Enumerator with the states of Process[Task, O] emitting data chunks of O. It’s quite simple actually.
Building Process1[I, O] from Iteratee[I, O]
The idea is to drive an Iteratee from a scalaz-stream Process so that it can consume an Enumerator and be used in Play WS.
An Iteratee[I, O] accepts inputs of type I (and nothing else) and will fold the input stream into a single result of type O.
A Process1[I, O] accepts inputs of type I and emits chunks of type O but not necessarily one single output chunk. So it’s a good candidate for our use-case but we need to choose which emitted chunk will be the result of the Iteratee[I, O]. here, totally arbitrarily, I’ve chosen to take the first emit as the result (but the last would be as good if not better).
So I implemented the following:
12345
defiterateeFirstEmit[I, O](p:Process.Process1[I, O])(implicitctx:ExecutionContext):Iteratee[I, O]={...// look the code in github project...}
The implementation is really raw for experimentation as it goes through the states of the Process1[I,O] and generates the corresponding states of Iteratee[I,O] until first emitted value. Nothing more nothing less…
A few basic action samples
Everything done in those samples could be done with Iteratee/Enumeratee more or less simply. The subject is not there!
Sample 1 : Generates a stream from a Simple Emitter Process
Sample 2 : Generates a stream from a continuous emitter
1234567
/** A process generating an infinite stream of natural numbers */valnumerals=Process.unfold(0){s=>valx=s+1;Some(x,x)}.repeat// we limit the number of outputs but you don't have it can stream forever...defsample2=Action{Ok.stream(enumerator(numerals.map(_.toString).intersperse(",").take(40)))}
Sample 3 : Generates a stream whose output frequency is controlled by a tee with numeral generator on left and ticker on right
1234567891011121314
/** ticks constant every delay milliseconds */defticker(constant:Int,delay:Long):Process[Task, Int]=Process.await(scalaFuture2scalazTask(delayedNumber(constant,delay)))(Process.emit).repeatdefsample3=Action{Ok.stream(enumerator(// creates a Tee outputting only numerals but consuming ticker // to have the delayed effect(numeralsteeticker(0,100))(processes.zipWith((a,b)=>a)).take(100).map(_.toString).intersperse(",")))}
Please note :
scalaFuture2scalazTask is just a helper to convert a Future into Task
tickeris quite simple to understand: it awaits Task[Int] and emits thisInt and repeats it again…
processes.zipWith((a,b) => a) is a tee (2 inputs left/right) that outputs only left data but consumes right also to have the delay effect.
.map(_.toString) simply converts into something writeable by Ok.stream
.intersperse(",") which simply add `”,” between each element
123456
>curl"localhost:10000/sample3"--no-buffer1...// to simulate the progressive apparition of numbers on screen1,...1,2......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
Sample 4 : Generates a stream using side-effect to control output frequency
12345678910111213141516171819
/** Async generates this Int after delay*/defdelayedNumber(i:Int,delay:Long):Future[Int]=play.api.libs.concurrent.Promise.timeout(i,delay)/** Creates a process generating an infinite stream natural numbers * every `delay milliseconds */defdelayedNumerals(delay:Long)={defstep(i:Int):Process[Task, Int]={Process.emit(i).then(Process.await(scalaFuture2scalazTask(delayedNumber(i+1,delay)))(step))}Process.await(scalaFuture2scalazTask(delayedNumber(0,delay)))(step)}defsample4=Action{Ok.stream(enumerator(delayedNumerals(100).take(100).map(_.toString).intersperse(",")))}
Please note:
delayedNumber uses an Akka scheduler to trigger our value after timeout
delayedNumerals shows a simple recursive `Process[Task, Int] construction which shouldn’t be too hard to understand
12345678
>curl"localhost:10000/sample4"--no-buffer0...// to simulate the progressive apparition of numbers every 100ms0,...0,1...0,1,...0,1,2......0,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
Sample 5 : Generates a stream by consuming completely another stream
12345678910111213141516171819
// a process folding all Array[Byte] into a big Stringvalreader:Process.Process1[Array[Byte], String]=processes.fold1[Array[Byte]]((a,b)=>a++b).map{arr=>newString(arr)}|>processes.lastdefsample5=Action{// the WS call with response consumer by previous Process1[Array[Byte], String] driving the Iteratee[Array[Byte], String]valmaybeValues:Future[String]=WS.url(routes.Application.sample2().absoluteURL()).get(rh=>iterateeFirstEmit(reader)).flatMap(_.run)Ok.stream(enumerator(// wraps the received String in a Process// re-splits it to remove ","// emits all chunksProcess.wrap(scalaFuture2scalazTask(maybeValues)).flatMap{values=>Process.emitAll(values.split(","))}))}
Please note:
reader is a Process1[Array[Byte], String] that folds all receivedArray[Byte]into aString`
iterateeFirstEmit(reader) simulates an Iteratee[Array[Byte], String] driven by the reader process that will fold all chunks of data received from WS call to routes.Application.sample2()
.get(rh => iterateeFirstEmit(reader)) will return a Future[Iteratee[Array[Byte], String] that is run in .flatMap(_.run) to return a Future[String]
Process.wrap(scalaFuture2scalazTask(maybeValues)) is a trick to wrap the folded Future[String] into a Process[Task, String]
Process.emitAll(values.split(",")) splits the resulting string again and emits all chunks outside (stupid, just for demo)
Building recursive streaming action consuming itself
Hacking WS to consume & re-emit WS in realtime
WS.executeStream(r: ResponseHeader => Iteratee[Array[Byte], A]) is cool API because you can build an iteratee from the ResponseHeader and then the iteratee will consume received `Array[Byte] chunks in a reactive way and will fold them. The problem is that until the iteratee has finished, you won’t have any result.
But I’d like to be able to receive chunks of data in realtime and re-emit them immediately so that I can inject them in realtime data flow processing. WS API doesn’t allow this so I decided to hack it a bit. I’ve written WSZ which provides the API:
123
defgetRealTime():Process[Future, Array[Byte]]// based onprivate[libs]defrealtimeStream:Process[Future, Array[Byte]]
This API outputs a realtime Stream of Array[Byte] whose flow is controlled by promises (Future) being redeemed in AsyncHttpClient AsyncHandler. I didn’t care about ResponseHeaders for this experimentation but it should be taken account in a more serious impl.
I obtain a Process[Future, Array[Byte]] streaming received chunks in realtime and I can then take advantage of the power of machines to manipulate the data chunks as I want.
Sample 6 : Generates a stream by forwarding/refolding another stream in realtime
/** A Process1 splitting input strings using splitter and re-grouping chunks */defsplitFold(splitter:String):Process.Process1[String, String]={// the recursive splitter / refolderdefgo(rest:String)(str:String):Process.Process1[String, String]={valsplitted=str.split(splitter)println(s"""$str - ${splitted.mkString(",")} --""")(splitted.lengthmatch{case0=>// string == splitter// emit rest// loopProcess.emit(rest).then(Process.await1[String].flatMap(go("")))case1=>// splitter not found in string // so waiting for next string// loop by adding current str to rest// but if we reach end of input, then we emit (rest+str) for last elementProcess.await1[String].flatMap(go(rest+str)).orElse(Process.emit(rest+str))case_=>// splitter found// emit rest + splitted.head// emit all splitted elements but last// loops with rest = splitted last elementProcess.emit(rest+splitted.head).then(Process.emitAll(splitted.tail.init)).then(Process.await1[String].flatMap(go(splitted.last)))})}// await1 simply means "await an input string and emits it"Process.await1[String].flatMap(go(""))}defsample6=Action{implicitrequest=>valp=WSZ.url(routes.Application.sample4().absoluteURL()).getRealTime.translate(Task2FutureNT)Ok.stream(enumerator(p.map(newString(_))|>splitFold(",")))}
Please note:
def splitFold(splitter: String): Process.Process1[String, String] is just a demo that coding a Process transducer isn’t so crazy… Look at comments in code
.translate(Task2FutureNF) converts the Process[Future, Array[Byte]] to Process[Task, Array[Byte]] using Scalaz Natural Transformation.
p |> splitFold(",") means “pipe output of process p to input of splitFold”.
Let’s finish our trip with a bit of puzzle and mystery.
THE FINAL MYSTERY: recursive stream generating Fibonacci series
As soon as my first experimentations of scalaz-stream with Play were operational, I’ve imagined an interesting case:
Is it possible to build an action generating a stream of data fed by itself: a kind of recursive stream.
With Iteratee, it’s not really possible since it can’t emit data before finishing iteration. It would certainly be possible with an Enumeratee but the API doesn’t exist and I find it much more obvious with scalaz-stream API!
The mystery isn’t in the answer to my question: YES it is possible!
The idea is simple:
Create a simple action
Create a first process emitting a few initialization data
Create a second process which consumes the WS calling my own action and re-emits the received chunks in realtime
Append first process output and second process output
Stream global output as a result of the action which will back-propagated along time to the action itself…
Naturally, if it consumes its own data, it will recall itself again and again and again until you reach the connections or opened file descriptors limit. As a consequence, you must limit the depth of recursion.
I performed different experiences to show this use-case by zipping the stream with itself, adding elements with themselves etc…
And after a few tries, I implemented the following code quite fortuitously :
/** @param curDepth the current recursion depth * @param maxDepth the max recursion depth */defsample7(curDepth:Int,maxDepth:Int)=Action{implicitrequest=>// initializes serie with 2 first numerals output with a delay of 100msvalinit:Process[Task, String]=delayedNumerals(100).take(2).map(_.toString)// Creates output Process// If didn't reach maxDepth, creates a process consuming my own action// If reach maxDepth, just emit 0valoutputProcess=if(curDepth<maxDepth){// calling my own action and streaming chunks using getRealTimevalmyself=WSZ.url(routes.Application.sample7(curDepth+1,maxDepth).absoluteURL()).getRealTime.translate(Task2FutureNT).map(newString(_))// splitFold isn't useful, just for demo|>splitFold(",")// THE IMPORTANT PART BEGIN// appends `init` output with `myself` output// pipe it through a helper provided scalaz-stream `processes.sum[Long]`// which sums elements and emits partial sums((initappendmyself).map(_.toLong)|>processes.sum[Long])// THE IMPORTANT PART END// just for output format.map(_.toString).intersperse(",")}elseProcess.emit(0).map(_.toString)Ok.stream(enumerator(outputProcess))}
I won’t tell the answer to this puzzling side-effect and let you think about it and discover why it works XD
But this sample shows exactly what I wanted: Yes, it’s possible to feed an action with its own feed! Victory!
Conclusion
Ok all of that was really funky but is it useful in real projects? I don’t really know yet but it provides a great proof of the very reactive character of scalaz-stream and Play too!
I tend to like scalaz-stream and I feel more comfortable, more natural using Process than Iteratee right now… Maybe this is just an impression so I’ll keep cautious about my conclusions for now…
All of this code is just experimental so be aware about it. If you like it and see that it could be useful, tell me so that we create a real library from it!
Have Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,
Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,
Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,Fun,!
Here are a few things that bother me when I use Play Iteratee (you don’t have to agree, this is very subjective):
Enumeratees are really powerful (maybe the most powerful part of the API) but they can be tricky: for ex, defining a new Enumeratee from scratch isn’t easy at first sight due to the signature of the Enumeratee itself, Enumeratee composes differently on left (with Enumerators) and on right (with Iteratees) and it can be strange at beginning…
Enumerators are not defined (when not using helpers) in terms of the data they produce but with respect to the way an Iteratee will consume the data they will produce. You must somewhat reverse your way of thinking which is not so natural.
Iteratees are great to produce one result by folding a stream of data but if you want to consume/cut/aggregate/re-emit the chunks, the code you write based on Iteratee/Enumeratee quickly becomes complex, hard to re-read and edge cases (error, end of stream) are hard to treat.
When you want to manipulate multiple streams together, zip/interleave them, you must write very complex code too.
End of iteration and Error management with Iteratees isn’t really clear IMHO and when you begin to compose Iteratees together, it becomes hard to know what will happen…
If you want to manipulate a stream with side-effecting, you can do it with Enumeratees but it’s not so obvious…
Now you should use play-autosource 2.0 correcting a few issues & introducing ActionBuilder from play2.2
The code for all autosources & sample apps can be found on Github here
Brand New Autosources
Play AutoSource now have 2 more implementations :
Datomic based on Datomisca, the Scala API I developed with Daniel James (@dwhjames) sponsored by Pellucid Analytics & Zenexity which is presented in this article
One month ago, I’ve demo’ed the concept of Autosource for Play2/Scala with ReactiveMongo in this article. ReactiveMongo was the perfect target for this idea because it accepts Json structures almost natively for both documents manipulation and queries.
But how does the concept behave when applied on a DB for which data are constrained by a schema and for which queries aren’t Json.
With ReactiveMongo Autosource, you could create a pure blob Autosource using JsObject without any supplementary information. But with Datomic, it’s not possible because Datomic forces to use a schema for your data.
We could create a schema and manipulate JsObject directly with Datomic and some Json validators. But I’m going to focus on the static models because this is the way people traditionally interact with a Schema-constrained DB.
Let’s create our model and schema.
123456789101112131415161718192021222324252627
// The Model (with characters pointing on Datomic named entities)caseclassPerson(name:String,age:Long,characters:Set[DRef])// The Schema written with DatomiscaobjectPerson{// Namespacesvalperson=newNamespace("person"){valcharacters=Namespace("person.characters")}// Attributesvalname=Attribute(person/"name",SchemaType.string,Cardinality.one).withDoc("Person's name")valage=Attribute(person/"age",SchemaType.long,Cardinality.one).withDoc("Person's age")valcharacters=Attribute(person/"characters",SchemaType.ref,Cardinality.many).withDoc("Person's characterS")// Characters named entitiesvalviolent=AddIdent(person.characters/"violent")valweak=AddIdent(person.characters/"weak")valclever=AddIdent(person.characters/"clever")valdumb=AddIdent(person.characters/"dumb")valstupid=AddIdent(person.characters/"stupid")// Schemavalschema=Seq(name,age,characters,violent,weak,clever,dumb,stupid)
Create Datomisca Autosource
Now that we have our schema, let’s write the autosource.
importdatomisca._importDatomic._importplay.autosource.datomisca._importplay.modules.datomisca._importImplicits._importscala.concurrent.ExecutionContext.Implicits.globalimportplay.api.Play.currentimportmodels._importPerson._objectPersonsextendsDatomiscaAutoSourceController[Person]{// gets the Datomic URI from application.confvaluri=DatomicPlugin.uri("mem")// ugly DB initialization ONLY for test purposeDatomic.createDatabase(uri)// Datomic connection is requiredoverrideimplicitvalconn=Datomic.connect(uri)// Datomic partition in which you store your entitiesoverridevalpartition=Partition.USER// more than ugly schema provisioning, ONLY for test purposeAwait.result(Datomic.transact(Person.schema),Duration("10 seconds"))}
Implementing Json <-> Person <-> Datomic transformers
If you compile previous code, you should have following error:
Actually, Datomisca Autosource requires 4 elements to work:
Json.Format[Person] to convert Person instances from/to Json (network interface)
EntityReader[Person] to convert Person instances from Datomic entities (Datomic interface)
PartialAddEntityWriter[Person] to convert Person instances to Datomic entities (Datomic interface)
Reads[PartialAddEntity] to convert Json to PartialAddEntity which is actually a simple map of fields/values to partially update an existing entity (one single field for ex).
It might seem more complicated than in ReactiveMongo but there is nothing different. The autosource converts Person from/to Json and then converts Person from/to Datomic structure ie PartialAddEntity. In ReactiveMongo, the only difference is that it understands Json so well that static model becomes unnecessary sometimes ;)…
Let’s define those elements in Person companion object.
objectPerson{...// Classic Play2 Json Reads/WritesimplicitvalpersonFormat=Json.format[Person]// Partial entity update : Json to PartialAddEntity ReadsimplicitvalpartialUpdate:Reads[PartialAddEntity]=(((__\'name).read(readAttr[String](Person.name))orElseReads.pure(PartialAddEntity(Map.empty)))and((__\'age).read(readAttr[Long](Person.age))orElseReads.pure(PartialAddEntity(Map.empty)))and// need to specify type because a ref/many can be a list of dref or entities so need to tell it explicitly(__\'characters).read(readAttr[Set[DRef]](Person.characters))reduce)// Entity Reads (looks like Json combinators but it's Datomisca combinators)implicitvalentity2Person:EntityReader[Person]=(name.read[String]andage.read[Long]andcharacters.read[Set[DRef]])(Person.apply_)// Entity Writes (looks like Json combinators but it's Datomisca combinators)implicitvalperson2Entity:PartialAddEntityWriter[Person]=(name.write[String]andage.write[Long]andcharacters.write[Set[DRef]])(DatomicMapping.unlift(Person.unapply))...}
Now we have everything to work except a few configurations.
Add AutoSource routes at beginning conf/routes
1
->/personcontrollers.Persons
Create conf/play.plugins to initialize Datomisca Plugin
1
400:play.modules.datomisca.DatomicPlugin
Append to conf/application.conf to initialize MongoDB connection
In Datomic, you can’t do a getAll without providing a Datomic Query.
But what is a Datomic query? It’s inspired by Datalog which uses predicates to express the constraints on the searched entities. You can combine predicates together.
With Datomisca Autosource, you can directly send datalog queries in the query parameter q for GET or in body for POST with one restriction: your query can’t accept input parameters and must return only the entity ID. For ex:
Please note the use of POST here instead of GET because Curl doesn’t like [] in URL even using -g option
Now you can use all other routes provided by Autosource
Autosource Standard Routes
Get / Find / Stream
GET /persons?… -> Find by query
GET /persons/ID -> Find by ID
GET /persons/stream -> Find by query & stream result by page
Insert / Batch / Find
POST /persons + BODY -> Insert
POST /persons/find + BODY -> find by query (when query is too complex to be in a GET)
POST /persons/batch + BODY -> batch insert (multiple)
Update / batch
PUT /persons/ID + BODY -> Update by ID
PUT /persons/ID/partial + BODY -> Update partially by ID
PUT /persons/batch -> batch update (multiple)
Delete / Batch
DELETE /persons/ID -> delete by ID
DELETE /persons/batch + BODY -> batch delete (multiple)
Conclusion
Play-Autosource’s ambition was to be DB agnostic (as much as possible) and showing that the concept can be applied to schemaless DB (ReactiveMongo & CouchDB) and schema DB (Datomic) is a good sign it can work. Naturally, there are a few more elements to provide for Datomic than in ReactiveMongo but it’s useful anyway.
Thank to @TrevorReznik for his contribution of CouchBase Autosource.
I hope to see soon one for Slick and a few more ;)
Do you remember JsPath pattern matching presented in this article ?
Let’s now go further with something that you should enjoy even more: Json Interpolation & Pattern Matching.
I’ve had the idea of these features for some time in my mind but let’s render unto Caesar what is Caesar’s : Rapture.io proved that it could be done quite easily and I must say I stole got inspired by a few implementation details from them! (specially the @inline implicit conversion for string interpolation class which is required due to a ValueClass limitation that should be removed in further Scala versions)
Please note that string variables must be put between "..." because without it the parser will complain.
Ok, so now it’s really trivial to write Json, isn’t it?
String interpolation just replaces the string you write in your code by some Scala code concatenating pieces of strings with variables as you would write yourself. Kind-of: s"toto ${v1} tata" => "toto + v1 + " tata" + ...
But at compile-time, it doesn’t compile your String into Json: the Json parsing is done at runtime with string interpolation. So using Json interpolation doesn’t provide you with compile-time type safety and parsing for now.
In the future, I may replace String interpolation by a real Macro which will also parse the string at compile-time. Meanwhile, if you want to rely on type-safety, go on using Json.obj / Json.arr API.
Json pattern matching
What is one of the first feature that you discover when learning Scala and that makes you say immediately: “Whoaa Cool feature”? Pattern Matching.
You can write:
12345678910111213
scala>valopt=Option("toto")opt:Option[String]=Some(toto)scala>optmatch{caseSome(s)=>s"not empty option:$s"caseNone=>"empty option"}res2:String=notemptyoption:toto// or direct variable assignement using pattern matchingscala>valSome(s)=opts:String=toto
30’ : Create new ReactiveMongo AutoSource Controller in app/Person.scala
123456789101112131415161718192021
packagecontrollersimportplay.api._importplay.api.mvc._// BORING IMPORTS// Jsonimportplay.api.libs.json._importplay.api.libs.functional.syntax._// Reactive JSONCollectionimportplay.modules.reactivemongo.json.collection.JSONCollection// Autosourceimportplay.autosource.reactivemongo._// AutoSource is Async so imports Scala Future implicitsimportscala.concurrent.ExecutionContext.Implicits.globalimportplay.api.Play.current// >>> THE IMPORTANT PART <<<objectPersonsextendsReactiveMongoAutoSourceController[JsObject]{valcoll=db.collection[JSONCollection]("persons")}
50’ : Add AutoSource routes at beginning conf/routes
1
->/personcontrollers.Persons
60’ : Create conf/play.plugins to initialize ReactiveMongo Plugin
With Play-Autosource, in a few lines, you obtain :
A backed abstract datasource (here implemented for ReactiveMongo but it could be implemented for other DBs)
All CRUD operations are exposed as pure REST services
The datasource is typesafe (here JsObject but we’ll show later that we can use any type)
It can be useful to kickstart any application in which you’re going to work iteratively on our data models in direct interaction with front-end.
It could also be useful to Frontend developers who need to bootstrap frontend code with Play Framework application backend. With Autosource, they don’t have to care about modelizing strictly a datasource on server-side and can dig into their client-side code quite quickly.
Adding constraints & validation
Now you tell me: “Hey that’s stupid, you store directly JsObject but my data are structured and must be validated before inserting them”
Yes you’re right so let’s add some type constraints on our data:
12345678910
objectPersonsextendsReactiveMongoAutoSourceController[JsObject]{valcoll=db.collection[JSONCollection]("persons")// we validate the received Json as JsObject because the autosource type is JsObject// and we add classic validations on typesoverridevalreader=__.read[JsObject]keepAnd((__\"name").read[String]and(__\"age").read[Int](Reads.min(0)keepAndReads.max(117))).tupled}
You can add progressively constraints on your data in a few lines. With AutoSource, you don’t need to determine immediately the exact shape of your models and you can work with JsObject directly as long as you need. Sometimes, you’ll even discover that you don’t even need a structured model and JsObject will be enough. (but I also advise to design a bit things before implementing ;))
Keep in mind that our sample is based on an implementation for ReactiveMongo so using Json is natural. For other DB, other data structure might be more idiomatic…
Use typesafe models
Now you tell me: “Funny but but but JsObject is evil because it’s not strict enough. I’m a OO developer (maybe abused by ORM gurus when I was young) and my models are case-classes…”
Yes you’re right, sometimes, you need more business logic or you want to separate concerns very strictly and your model will be shaped as case-classes.
So let’s replace our nice little JsObject by a more serious case class.
1234567891011
// the modelcaseclassPerson(name:String,age:Int)objectPerson{// the famous Json Macro which generates at compile-time a Reads[Person] in a one-linerimplicitvalfmt=Json.format[Person]}// The autosource... shorter than beforeobjectPersonsextendsReactiveMongoAutoSourceController[Person]{valcoll=db.collection[JSONCollection]("persons")}
Please note that I removed the validations I had introduced before because there are not useful anymore: using Json macros, I created an implicit Format[Person] which is used implicitly by AutoSource.
So, now you can see why I consider AutoSource as a typesafe datasource.
Let’s be front-sexy with AngularJS
You all know that AngularJS is the new kid on the block and that you must use it if you want to be sexy nowadays.
I’m already sexy so I must be able to use it without understanding anything to it and that’s exactly what I’ve done: in 30mn without knowing anything about Angular (but a few concepts), I wrote a dumb CRUD front page plugged on my wonderful AutoSource.
Client DS in app/assets/javascripts/persons.js
This is the most important part of this sample: we need to call our CRUD autosource endpoints from angularJS.
We are going to use Angular resources for it even if it’s not really the best feature of AngularJS. Anyway, in a few lines, it works pretty well in my raw case.
(thanks to Paul Dijou for reviewing this code because I repeat I don’t know angularJS at all and I wrote this in 20mn without trying to understand anything :D)
varapp=// injects ngResourceangular.module("app",["ngResource"])// creates the Person factory backed by our autosource// Please remark the url person/:id which will use transparently our CRUD AutoSource endpoints.factory('Person',["$resource",function($resource){return$resource('person/:id',{"id":"@id"});}])// creates a controller.controller("PersonCtrl",["$scope","Person",function($scope,Person){$scope.createForm={};// retrieves all persons$scope.persons=Person.query();// creates a person using createForm and refreshes list$scope.create=function(){varperson=newPerson({name:$scope.createForm.name,age:$scope.createForm.age});person.$save(function(){$scope.createForm={};$scope.persons=Person.query();})}// removes a person and refreshes list$scope.remove=function(person){person.$remove(function(){$scope.persons=Person.query();})}// updates a person and refreshes list$scope.update=function(person){person.$save(function(){$scope.persons=Person.query();})}}]);
CRUD UI in index.scala.html
Now let’s create our CRUD UI page using angular directives. We need to be able to:
list persons
update/delete each person
create new persons
123456789101112131415161718192021222324
@(message: String)
@main("Welcome to Play 2.1") {
<divng-controller="PersonCtrl"><!-- create form --><labelfor="name">name:</label><inputng-model="createForm.name"/><labelfor="age">age:</label><inputng-model="createForm.age"type="number"/><buttonng-click="create()">Create new person</button><hr/><!-- List of persons with update/delete buttons --><table><thead><th>name</th><th>age</th><td>actions</td></thead><tbodyng-repeat="person in persons"><tr><td><inputng-model="person.name"/></td><td><inputtype="number"ng-model="person.age"/></td><td><buttonng-click="update(person)">Update</button><buttonng-click="remove(person)">Delete</button></td></tr></tbody></div></div>}
Import Angular in main.scala.html
We need to import angularjs in our application and create angular application using ng-app
I know what you think: “Uhuh, the poor guy who exposes his DB directly on the network and who is able to delete everything without any security”
Once again, you’re right. (yes I know I love flattery)
Autosource is by default not secured in any way and actually I don’t really care about security because this is your job to secure your exposed APIs and there are so many ways to secure services that I prefer to let you choose the one you want.
Anyway, I’m a nice boy and I’m going to show you how you could secure the DELETE endpoint using the authentication action composition sample given in Play Framework documentation.
// FAKE USER class to simulate a user extracted from DB.caseclassUser(name:String)objectUser{deffind(name:String)=Some(User(name))}objectPersonsextendsReactiveMongoAutoSourceController[Person]{// The action composite directly copied for PlayFramework docdefAuthenticated(action:User=>EssentialAction):EssentialAction={// Let's define a helper function to retrieve a UserdefgetUser(request:RequestHeader):Option[User]={request.session.get("user").flatMap(u=>User.find(u))}// Now let's define the new ActionEssentialAction{request=>getUser(request).map(u=>action(u)(request)).getOrElse{Done(Unauthorized)}}}valcoll=db.collection[JSONCollection]("persons")// >>> IMPORTANT PART <<<// We simply override the delete action// If authenticated, we call the original actionoverridedefdelete(id:BSONObjectID)=Authenticated{_=>super.delete(id)}defindex=Action{Ok(views.html.index("ok"))}// the login action which log any userdeflogin(name:String)=Action{Ok("logged in").withSession("user"->name)}// the logout action which log out any userdeflogout=Action{Ok("logged out").withNewSession}}
Nothing to complicated here.
If you need to add headers in your responses and params to querystring, it’s easy to wrap autosource actions. Please refer to Play Framework doc for more info…
I won’t try it here, the article is already too long but it should work…
Play-Autosource is DB agnostic
Play-Autosource Core is independent of the DB and provides Reactive (Async/Nonblocking) APIs to fulfill PlayFramework requirements.
Naturally this 1st implementation uses ReactiveMongo which is one of the best sample of DB reactive driver. MongoDB fits very well in this concept too because document records are really compliant to JSON datasources.
But other implementations for other DB can be done and I count on you people to contribute them.
DB implementation contributions are welcome (Play-Autosource is just Apache2 licensed) and AutoSource API are subject to evolutions if they appear to be erroneous.
Conclusion
Play-Autosource provides a very fast & lightweight way to create a REST CRUD typesafe datasource in your Play/Scala application. You can begin with blob data such as JsObject and then elaborate the model of your data progressively by adding constraints or types to it.
There would be many more things to say about Play/Autosource:
you can also override writers to change output format
you have some alpha streaming API also
etc…
There are also lots of features to improve/add because it’s still a very draft module.
If you like it and have ideas, don’t hesitate to discuss, to contribute, to improve etc…
curl -X POST -d "{ "coding" : "Have fun"} http://localhost:9000/developer
Now, you may certainly have realized I’m Play2.1 Json API advocate. But you may also have understood that I’m not interested in Json as an end in itself. What catches my attention is that it’s a versatile arborescent data structure that can be used in web server&client, in DB such as ReactiveMongo and also when communicating between servers with WebServices.
So I keep exploring what can be done with Json (specially in the context of PlayFramework reactive architecture) and building the tools that are required to concretize my ideas.
My last article introduced JsPath Pattern Matching and I told you that I needed this tool to use it with JsZipper. It’s time to use it…
Here is why I want to do:
Build dynamically a Json structure by aggregating data obtained by calling several external WS such as twitterAPI or github API or whatever API.
Build this structure from a Json template stored in MongoDB in which I will find the URL and params of WebServices to call.
Use Play2.1/WS & ReactiveMongo reactive API meaning resulting Json should be built in an asynchronous and non-blocking way.
Use concept of JsZipper introduced in my previous article to be able to modify efficiently Play2.1/Json immutable structures.
Please note that this idea and its implementation is just an exercise of style to study the idea and introduce technical concepts but naturally it might seem a bit fake. Moreover, keep in mind, JsZipper API is still draft…
The idea of Json template
Imagine I want to gather twitter user timeline and github user profile in a single Json object.
I also would like to:
configure the URL of WS and query parameters to fetch data
Using the url and user_id found in __\streams\twitter, I can call twitter API to fetch the stream of tweets and the same for__\streams\github`. Finally I replace the content of each node as following:
12345678910
{"streams":{"twitter":{// TWITTER USER TIMELINE HERE},"github":{// GITHUB USER PROFILE HERE}}}
Moreover, I’d like to store multiple templates like previous sample with multiple user_id to be able to retrieve multiple streams at the same time.
Creating Json template in Play/ReactiveMongo (v0.9)
Recently, Stephane Godbillon has released ReactiveMongo v0.9 with corresponding Play plugin. This version really improves and eases the way you can manipulate Json directly with Play & Mongo from Scala.
Let’s store a few instance of previous templates using this API:
12345678910111213141516171819202122232425
// gets my mongo collectiondefcoll=db.collection[JSONCollection]("templates")defprovision=Action{Async{valvalues=Enumerator(Json.obj("streams"->Json.obj("twitter"->Json.obj("url"->"http://localhost:9000/twitter/statuses/user_timeline","user_id"->"twitter_nick1"),"github"->Json.obj("url"->"http://localhost:9000/github/users","user_id"->"github_nick1"))),...moretemplates)coll.bulkInsert(values).map{nb=>Ok(Json.obj("nb"->nb))}}}
Hard isn’t it?
Note that I use localhost URL because with real Twitter/Github API I would need OAuth2 tokens and this would be a pain for this sample :)
Reactive Json crafting
Now, let’s do the real job i.e the following steps:
retrieve the template(s) from Mongo using ReactiveMongo JsonCollection
call the WebServices to fetch the data using Play Async WS
update the Json template(s) using Monadic JsZipper JsZipperM[Future]
The interesting technical points here are that:
ReactiveMongo is async so we get Future[JsValue]
Play/WS is Async so we get also Future[JsValue]
We need to call multiple WS so we have a Seq[Future[JsValue]]
We could use Play/Json transformers presented in a previous article but knowing that you have to manage Futures and multiple WS calls, it would create quite complicated code.
Here is where Monadic JsZipper becomes interesting:
JsZipper allows modifying immutable JsValue which is already cool
JsZipperM[Future] allows modifying JsValue in the future and it’s even better!
Actually the real power of JsZipper (besides being able to modify/delete/create a node in immutable Json tree) is to transform a Json tree into a Stream of nodes that it can traverse in depth, in width or whatever you need.
Less code with WS sequential calls
Here is the code because you’ll see how easy it is:
12345678910111213141516171819
// a helper to call WSdefcallWSFromTemplate(value:JsValue):Future[JsValue]=WS.url((value\"url").as[String]).withQueryString("user_id"->(value\"user_id").as[String]).get().map{resp=>resp.json}// calling WS sequentiallydefdataSeq=Action{Async{for{templates<-coll.find(Json.obj()).cursor[JsObject].toList// retrieves templates from Mongoupdated<-Json.toJson(templates).updateAllM{case(_\"twitter",value)=>callWSFromTemplate(value)case(_\"github",value)=>callWSFromTemplate(value)case(_,value)=>Future.successful(value)}}yieldOk(updated)}}
Please note:
Json.toJson(templates) transforms a List[JsObject] into JsArray because we want to manipulate pure JsValue with JsZipperM[Future].
.updateAllM( (JsPath, JsValue) => Future[JsValue] ) is a wrapper API hiding the construction of a JsZipperM[Future]: once built, the `JsZipperM[Future] traverses the Json tree and for each node, it calls the provided function flatMapping on Futures before going to next node. This makes the calls to WS sequential and not parallel.
case (_ \ "twitter", value) : yes here is the JsPath pattern matching and imagine the crazy stuff you can do mixing Json traversal and pattern matching ;)
Async means the embedded code will return Future[Result] but remember that it DOESN’T mean the Action is synchronous/blocking because in Play, everything is Asynchronous/non-blocking by default.
Then you could tell me that this is cool but the WS are not called in parallel but sequentially. Yes it’s true but imagine that it’s less than 10 lines of code and could even be reduced. Yet, here is the parallelized version…
Parallel WS calls
1234567891011121314151617181920
defdataPar=Action{Async{coll.find(Json.obj()).cursor[JsObject].toList.flatMap{templates=>// converts List[JsObject] into JsArrayvaljsonTemplates=Json.toJson(templates)// gathers all nodes that need to be updatedvalnodes=jsonTemplates.findAll{case(_\"twitter",_)|(_\"github",_)=>truecase(_,value)=>false}// launches WS calls in parallel and updates original JsArrayFuture.traverse(nodes){case(path@(_\"twitter"),value)=>callWSFromTemplate(value).map(resp=>path->resp)case(path@(_\"github"),value)=>callWSFromTemplate(value).map(resp=>path->resp)}.map{pathvalues=>Ok(jsonTemplates.set(pathvalues:_*))}}}}
Note that:
jsonTemplates.findAll( filter: (JsPath, JsValue) => Boolean ) traverses the Json tree and returns a Stream[(JsPath, JsValue)] containing the filtered nodes. This is not done with Future because we want to get all nodes now to be able to launch all WS calls in parallel.
Future.traverse(nodes)(T => Future[T]) traverses the filtered values and calls all WS in parallel.
case (path@(_ \ "twitter"), value) is just JsPath pattern matching once again keeping track of full path to be able to return it with the value path -> resp for next point.
jsonTemplates.set( (JsPath, JsValue)* ) finally updates all values at given path. Note how easy it is to update multiple values at multiple paths.
A bit less elegant than the sequential case but not so much.
Conclusion
This sample is a bit stupid but you can see the potential of mixing those different tools together.
Alone, JsZipper and JsPath pattern matching provides very powerful ways of manipulating Json that Reads/Writes can’t do easily.
When you add reactive API on top of that, JsZipper becomes really interesting and elegant.
While experimenting Play21/Json Zipper in my previous article, I needed to match patterns on JsPath and decided to explore a bit this topic.
This article just presents my experimentations on JsPath pattern matching so that people interested in the topic can tell me if they like it or not and what they would add or remove. So don’t hesitate to let comments about it.
If the result is satisfying, I’ll propose it to Play team ;)
Note the \?\ operator which is also a temporary choice: I didn’t want to choose \\ ause \?\ operator only works in the case where you match between the first and the last element of the path and not between anything and anything…
JsZipper is a new tool allowing much more complex & powerful manipulations of Json structures for Play2/Json Scala API (not a part of Play2 core for now)
JsZipper is inspired by the Zipper concept introduced by Gérard Huet in 1997.
The Zipper allows to update immutable traversable structures in an efficient way. Json is an immutable AST so it fits well. FYI, the Zipper behaves like a loupe that walks through each node of the AST (left/right/up/down) while keeping aware of the nodes on its left, its right and its upper. The interesting idea behind the loupe is that when it targets a node, it can modify and even delete the focused node. The analogy to the pants zipper is quite good too because when it goes down the tree, it behaves as if it was opening the tree to be able to drive the loupe through all nodes and when it goes up, it closes back the tree… I won’t tell more here, it would be too long.
JsZipper is a specific interpretation of Zipper concept for Play/Json API based on :
Scala Streams to go through / update / construct Json AST nodes in a lazy way
Monadic aspects to provide funnier ways of manipulating the Json AST (plz see below)
Please note, JsZipper is not an end in itself but a tool useful to provide new API to manipulate Json.
Let’s go to samples because it explains everything.
Let’s use Future as our Monad because it’s… coooool to do things in the future ;)
Imagine you call several services returning Future[JsValue] and you want to build/update a JsObject from it.
Until now, if you wanted to do that with Play2/Json, it was quite tricky and required some code.
It’s still draft so it can be improved but if you like it, don’t hesitate to comment and if people like it, it could become a part of Play Framework itself