Compositional Streaming with FS2

scala.bythebay.io: Michael Pilquist, Compositional Streaming with FS2

Recording: scala.bythebay.io: Michael Pilquist, Compositional Streaming with FS2

so I'm really excited to be here like Bill vinners my laptop has a bit of a slant on the front so I have to stand here and hold it to make sure it doesn't fall off this stand but I'm gonna be speaking about the project formerly known as scholar Z stream today now known as functional streams for Scala or FS 2 for short so I'm really excited to be here and I'm excited because this is over we did we did a bunch of redesign work in scalzi stream and specifically there was some you know really nasty resource allocation problems in light of like asynchronous operations where you know sort of our killer feature if you will of making sure resources are cleaned up no matter what type of thing occurred was failing in various cases right and if it's like our big killer feature and we can you know deliver on it it's pretty embarrassing so you know in the summer of 2015 Paul chiisana started but we thought would be a mild redesign work we were thinking maybe a month or two or thuh you know additions to the core algebra of scalzi stream and you know where we ended up about 15 months later in September was brand new library and it really is a brand new library I don't think there's a line of code that's shared between the two systems so you know for today's talk I started the option of either doing a brief introduction to the library and sort of getting you started or sort of showing all the fancy things we can do with it we're doing both but fortunately for you alexei only gave me 40 minutes so so we're gonna do the intro before we get started I do want to kind of contrast though zero at eight and zero at nine so series you're about eight you know it's in the git repo here it is actively maintained you know we just recently published zero eight six we published it for Scala two ten eleven and twelve it's currently published for Scala or z71 in seven two and you know we will publish a version for seven three as soon as that's out of milestones or release candidates you know what we do to have some dependencies they're certainly right the library does depend on Scala Z hence its name and it has dependency ons codec bits for some binary manipulation and and you know it is actively maintained we do that because we know the the jump from scalzi stream to fs2 is big right everybody has the appropriate time in their own business when it makes sense to take that take the time out to to do this transition so we are actively maintaining this you know we're open to adding some features but if you want to add like something huge you know talk to me but but series 0.9 is what we released a few months ago and it's quite a bit different because the FP landscape in Scala has changed quite a bit in the last two years we do have no dependencies right so you can sort of guess like anytime someone says they have no dependencies that really means they're reinventing everything themselves we did that we do have support for Scala Jay yes so you can do like full FP you know resource based streaming in the browser and we have bindings for lots of different you know libraries in the community we do have scalzi and cats bindings and we also have it bindings to a bunch of other things I'll talk about a bit you know most importantly our library is correct right the whole goal of the 0.8 to 0.9 transition was really making sure resource allocation was bulletproof but there was no type of strange edge case where if you did something you know in the you know in a very specific set of interleavings of processor execution we had missed some resource allocation free so we did that by defining this very small core interpreter and when you look at like the stream API that I'm gonna show today we're just looking at the surface API around a small core interpreter to give you an idea the the internal stream core interpreter that's in the guts of fs2 is nine primitive operations and everything else in the library is built from those nine primitive operations and in a very combinatorial explosion type type way right so the primary type in the library is this stream type for folks towards Scalzi stream this is analogous to process and it's parameterize by two type parameters right we have this F type and we have an output Type O and so we think of a stream we think about getting values out of it all right pooling values from it and that's our type o there our type F is some type affect that we're going to evaluate potentially um while pulling elements from this stream and we might evaluate that effect for various reasons right we might evaluate effect because we want to generate a no value but we might also evaluate an effect just for the purposes of the effect alright so a really good example of this I think is that like you want to open a socket to a server right we're going to evaluate some effect that opens a socket and there's no output to that you know necessarily or real if we're gonna close a socket or open a file or close a file all these effects were evaluating for the purpose of what they're doing to the outside world right but then other effects you know like reading bytes from a file or reading bytes from a from a socket is going to produce some output value and that's our output of typos so let's look at you know constructing a stream via the stream can apply you know the stream Campania to apply so here we have this Val s and we've you know created a stream with three elements it's discrete you know stream and I think what's interesting about this example is that the effect type here is set to nothing right so anytime we set that effect type to nothing what we're saying is we have a stream that cannot evaluate effects okay now later we could come in and say well but I do want to read from a file or I do want to write to a file we can sort of you know fix that up but we use this nothing type to indicate that this particular stream is not going to evaluate any effects and as you can imagine you know the stream API provides this sort of collection like API we can map we can append right we can do all sorts of things and in this particular case we can call this - list method and so if our stream has no effects right if we have that nothing type for our effect type then we can convert our stream to a strict collection right and kind of go back to the collection API we also have you know other sort of collection like api's that don't exist in standards collection library intersperse is one my favorites because it's super useful right but like stick is zero between elements right and again we can call to lists because because we have a pure stream and that's where the way I want you to think about this notion of having like nothing for an effect type is that it's a stream that's pure right because we have no effect evaluation occurring you know this is the to do MVC app of streaming libraries or the fizzbuzz of streaming libraries right the Fibonacci sequence and they're the reason we we use the Fibonacci sequence is because it does you know show a certain number of properties about a streaming library you know here we have this fib stream you know it's a pure stream because to the effect type is nothing we start with a you know stream of two elements 0 & 1 and then we append to that Phibbs zipped with its tail right and we combine the two elements like you know the element from fib the albums from fibs tale using just integer addition and that's infinite of course right so we've got some interesting properties we can sort of guess here there's some infinite behavior going on there's something that's handling like you know making sure we don't overflow the stack and then we can you know take the first ten elements and again convert back to a list and we have the Fibonacci sequence unlike the built-in stream type in the Scala Center collections we do not do any memoization so like this will be snappy even on this old MacBook Air if you said take 20 elements you know go get coffee right so this is not necessarily the best way to define a Fibonacci calculation but nonetheless it does show some interesting properties other streams how many people have used tasks okay about half so I want to look at effectual streams and in order to talk about effective streams we have to talk about what it means to evaluate an effect so here I have this task of sample the idea here is that sample the Val sample describes a program that when we evaluate it it's going to go off and like read some hardware sensor alright maybe it reads like the amount of heat that's it's being used by the JVM or read something off of a PCI bus or something so every time this sample is evaluated we're going to call this sensor dot read method and return the result okay and we can stick that into a stream using this eval constructor right we can call stream dot eval and what that will yield us is a stream of tasks on a sample right so we've sort of like taken apart the task sample and sort of moved those two types position to the stream type constructor now if we just say stream that eval sample we get a singleton stream a stream with one element and so we repeat it right and repeat we'll just infinitely you know continuously generate values and then we might want to do something with this so let's say in this particular example we take the first ten elements from the stream and by taking ten we've we've taken this infinite stream and turn it into a finite stream right one that as someone pulls from it once they pulled up to ten individual elements then halt even though the source was infinite and then in this case we use this run log method which is sort of like a sort of like a sink we're sort of like a terminating point where it's gonna pull elements into a vector right and it's gonna keep pulling until you reach the natural end of the stream so we get back in this particular case is like a task of vector of sample since we're still in our task constructor all we really have done is sort of taken a Streamy description of the program we wanted to build and turn it into an effect full description of the program so nothing has occurred at this point and sort of like a compilation step right and then finally somewhere towards like the end of the universe we can call this unsafe run method to say like now I want to go do my effects right and in fs2 all throughout anywhere you see a method starts with unsafe you should close your eyes you know in general these are the places where things happen right nothing else happens until that and you don't worry about the code on this slide but there are a bunch of different ways to run effectual streams as sort of the point here we've got like four four ways built into the library and if you use some of the Interop libraries there's some additional ways you know they all start with run and they all require something about our type constructor something about our about our F for our effect and specifically they require this instance of this catchable type class which I'll talk about more in a minute but because of this if we had a pure stream we can't call like run log because there's no catchable instance from nothing right so you can sort of think of like pure streams and effectual streams with pure stream that can always call to lists to vector and on effect will streams I have these run methods okay so once we have streams the next sort of most normal thing we want to do is transform the elements of streams and we do that with pipes okay so pipes are kind of interesting in that we have input values of type I we have output values of type O and we can also evaluate effects while we're transforming things okay so it's a little bit different than maybe in for folks that are familiar with us ecology stream with process one we weren't able to evaluate effects we had a different type to do that but here we have this disability to evaluate effects and this this is maybe an example of what stream evaluation might or train stream transpiration might look like we have two pipes defined here utf-8 decode and lines beauty f/8 the code pipe is going to pull some bytes in right from some source stream and turn them into strings using some utf-8 library and lines is gonna pull strings in and output strings but maybe like re chunk the the strings based off of a carriage return stream and then you know in this example down below we can do something like read all of the bytes from a file in like four K chunks and then use this through Combinator to say take the output of that original source put it through the utf-8 decoder and then put it through the text dot lines you know reparse ER and so this is a you know the the way we will go about doing stream transformations throughout the library we can also combine streams right not just transfer them but combine them so here we have pipe two which says I want to have two different inputs I've got a stream that has some output of type I and another stream that has an output of i2 and I want to hook those both together and create a new stream that outputs something Type O and again I can evaluate effects when I do this right maybe look at the values and go off and read a socket or something okay and we've seen an example of pipe two's already because we saw this zip with thing right here we had two streams we had fibs and we had pizzas tale those are two distinct input streams and we combined them via this addition method now you know this is pretty syntax on top of something that's pretty terrible we do have this through to Combinator through to is like the pipe to equivalent up through and maybe it's not so bad you can see there's like this zip with pipe too on the far right so there's some object called pipe too that gives us instances of the pipe to type right then get this like funky pure thing which is to avoid some inference things so don't worry too much about it we provide nice syntax for all this but you know we have we have seen this example of combining streams already and there's a bunch of different ways to do this so like in this particular case we're using pure stream so you know x and y are both discrete pure streams and we just inner leave the elements and again you know that's dressing on top of this uglier through two application okay so besides transforming and combining we also might want to sync streams put the value somewhere you know go do something with them and we have this sink type so again we have like an input type I but we don't have any outputs right and we can evaluate effects of type app so a good example of a sink might be writing to a file right in this particular case we're using like the i/o package from fs2 and we create a sink that writes all bytes that are sent to it into my file inside of the task constructor okay inside of the task type constructor and we can apply sinks two streams via the two Combinator so here we have like some you know stream of lines some some stream of tasks comma string and we just send the output of that to our output file and of course you know nothing has actually occurred yet we've built up this stream e description of what's going to occur and we sort of compile that or reinterpret that into an effect full description of what's going to occur and then eventually at some point you know run that program we saw Jay talked about caca you know last talk we do Kafka you know you can do Kafka stuff with sink so you can imagine like consumers from a Kafka topic as streams and you can imagine writing maybe events to Kopke you know using sinks right right all the events to a Kafka topic with a sink and this is sort of kind of taken from some some of our code at work we basically go and create some sync that represents a specific data type being written to Kafka and maybe we like map some elements and then eventually sync that the source stream into a Kafka topic again you know running it at the end of of the world so this is sort of my favorite part of fs2 so how do we compose these things together we have pipes you've got sinks we've got you know pipe twos we've got streams you know we've heard a lot this week that like the ultimate way to evaluate a library is to see how well it composes right so here's something I really like about FS - a pipe is really just a type alias for a function one okay so this is really meaningful this is this is an interesting thing about the library a pipe FIO you know we have an input of stream fi and it returns an output of stream F oh so for anyone familiar with Scalzi stream and the way process ones were defined it was very much different right process ones were like sort of a special type of machine I could sort of kind of like pull elements from places and was really confusing and I think like three people in the world understood it function ones everybody sort of guessed right and similarly pipe twos are just function twos and sinks are just pipes whose output type is set to unit okay so this means it's really easy to create your own pipes your own sinks and your own pipe twos right because these are just functions and of course we can do like fancy function composition with you know existing operators you know here we're just saying text utf-8 to code and then go to lines right it's just function composition and so a lot of times you'll see like people come to the library and say like oh I'm really stuck I want to figure out how to put you know this thing into that thing and they forget that like a pipe is really just this function one alias and as soon as they remember that like oh no I know how to do this right it's a really sort of important point in a NFS - composition how many people have done the readme example in FS 2's get repo okay so some so good I was afraid like everyone was raised their hands and we'd have to skip this but let you know when Paul T Asano first sort of announce Scalzi stream it like any Scala in 2012 maybe he did this example and this example now has been ported to FS 2 and I want to take a look at what it looks like the general idea is that we've got some file on disk called Fahrenheit X that we want to read it in and convert it to a new file in disk called Celsius text and like all big data shops this is going to be a 10 gigabyte file on disk and we need to do all of these conversions very quickly in constant keep so bear with me so the way we do that here is we use again this iope akka JEP s2 and as we say read all of the bytes from this Fahrenheit txt file in 4 kilobyte chunks inside the effect task we then want to put it push it through push those bytes through that pipe we just composed all right so we take this utf-8 decode do the line chunkiness and then we just want to start using some collection like operation so here we get rid of like all empty lines we get rid of all lines that start with comments we map over each individual line and parse the string as a double convert the double from Fahrenheit to Celsius then convert it back to a string and you might think like woah that you know what if the line is like asdf it's gonna blow open you know blow up it'll throw an exception and that's fine like we catch that exception we handle it appropriately inside the stream algebra right so you don't necessarily have to switch to you know returning like ethers or tries or something we do handle exceptions thrown from pure functions in the library now that we've had like this stream of Celsius values we want to stick new lines back in between all the strings regan code it's utf-8 write it to some file on disk and then run that overall program and now what we've got back is this converter value whose output type is meaningless right it's a task that says when we go and evaluate this we're gonna do this conversion in constant memory we're gonna be pulling elements as fast as we can write them to disk right so you can have a 10 gig file and this all works but it doesn't have an output value right that we have a task of unit which is the result of calling us around at the end which is to say that this this program this effect full value that we've evaluated here exists solely for the purpose of doing this stream transformation on disk we don't care about the output value and at some point we go off and we run that and we actually go to disk and do our i/o ok that's where it makes sense alright so we had resources you know being allocated there we had you know an input resource like a file handle we were reading from we had a file handle we were writing to and we do this via this bracket method so don't don't worry too much about the signature there but the general idea is that we're gonna have some effect full value that tells us how to go get a resource like go get a file handle or go to go open a socket and then once we've done that once we've successfully opened a socket or gotten a file we want to generate some stream from it like pulling bytes from it and we'll also give it some callback to tell it how to close that thing at the appropriate time and what we're gonna do is sort of decorate the stream that comes back with resource allocation and management logic such that if the stream ends in any way we're guaranteed that our release action will be called at appropriate time and like the release action sort of needs to be called at the most opportunistic time right you can't just say well when the JVM exits like well sort of you know build up all of these individual release calls and eventually get around to calling them we want to sort of get to it as soon as we can right even in light of like streams that go off and evaluate things asynchronously and it starts to get to like the core issue that fs2 addresses so it's worth pointing out that there's like multiple ways of stream can terminate and there's basically three we can reach the end right we've seen a lot of finite streams like stream 1 2 3 and so in the when we pull the last element out of stream 1 2 3 have reached a natural end but we also saw like infinite streams where we did like take 10 right and in those types of streams we have like this early termination that occurs where some Combinator some pipe decides to stop reading no don't don't pull any more elements and then finally we have like errors that occur we might want to uh you know terminate overall execution from and so even in light of all of those different types of errors we need some way that this research allocation can occur and the way we do that is with this pull data type so the general idea is that a stream can be opened so a stream fa for any fa can be opened and we get a pull of fa our and pools going to let us evaluate effects it's going to let us output values of type a and then it sort of lets us pass around this value of type R this resource of type r poll can also acquire resources like going out and getting a file handle or going out and opening a socket and so when we open a stream we're in this pull effect now and pull forms this monad in our R type and so what we can do is decide like whether we want to pull elements from the source stream that we opened or maybe just want to output things anyway maybe want to output two of everything we get but eventually we just we we did decide what type of things we're gonna do with the source stream and then we close it and we get back to a stream again maybe a modified stream a stream that looks like a some transformation on the original yeah Joe very far I think I didn't notice that that's good that's good so more specifically when we open a stream our R type our resource type is a handle okay and a handle is a is like a is literally a handle for the source stream that we can pull elements from this is really important point you might have noticed it as I've been speaking fs2 is completely pull based all right there's no pushing that occurs and as there is a lot of that like back pressure is just natural it's baked in right or maybe back pressures even like a type error to apply to this system because it makes no sense to talk about back pressure in something that pulls now we do have ways like if you want to do like push type mechanisms there are ways to do that with us too in a back pressure II sort of way but the the general notion here is that we get these handles and these handles let us pull elements from the source stream and you can see the handle is sort of sitting in monadic point of our poll datatype right so here's a way we might use this pull handle structure to implement head right so this is a Combinator that would just get the first element from a stream and terminate and the idea is that we would take some source stream and open it and then we would call flat map on it so now we're inside of the pull effect and what we get is our R type we get this handle H and with this handle H will say pull one element receive one element from the upstream from our source okay that gives us a tuple it gives us the single element that we asked for oh and it gives us like a continuation sort of a new handle if we want to read more elements and we don't care about that here because we just want to output the first so we use this output one constructor to say I'll put the 100 we got and then close the overall stream and what we end up with is a non-blocking implementation of the head Combinator that works for all types of streams regardless of where they're coming from regardless of what type of um you know asynchronous behavior that exists in that stream which guarantees resource safe you okay so sort of like a little DSL if you will you can imagine some nice free algebra sitting under underneath of this which is exactly what does sit underneath of this you know similarly we can implement tail and it's like the same exact pattern but we ignore the first value and we take that next H and we just echo it right so the H dot echo is sort of this built-in convenience operator that just says start pulling elements and outputting everything we get right so we throw away the first value and and echo so anybody defined like custom graph stages for acha streams right this is sort of what we're doing here the equivalent of like a custom graph stage so it's all type checked and you get very nice very nice compiler errors if you get it wrong and this pattern of like go and open a stream get into the pull effect run some function that reads a handle right and and returns a pull and then close again is so common that we have this source dot pole alias for it so think about the type of F in source dot pool right F is a function from handle to pull because we're going to do something with a handle and return a new pole that has you're outputted elements or you know decides how we're going to terminate the stream something to that effect and so what you can start building up our nice recursive poles we can say things like take this is a reasonable definition of take and fs2 and we'll define it recursively will say take works for any effect F and any type a completely unconstrained right we know it's unconstrained because we've provided no constraints on on these types right and we're gonna return this pipe FAA and at the very bottom of our body of our method we're going to you know just just materialize some source right could remember our pipes just a function one so we're gonna materialize this source stream source and we're gonna pull something from it and we're gonna replace these question marks with some function from handle to pull and so we'll define this loop as a recursive function right so loop will have state like the number of elements remaining and what it's going to return is this function from handle to pull okay and then inside the function we can just say if you know remaining is less than equal to zero we're done you know stop pulling closed the overall stream and otherwise receive a single element from our source we get like that single element a and we get a new handle which we shadow our old handle output the a and then go and loop again right so you can imagine there's some type of you know stack safety here it's occurring to make sure all of these binds all these flat map calls end up getting a trampling and that greater than greater then is just the indent operator it's like flat map that ignores its input value and we can sort of do this in all sorts of interesting ways right the same basic pattern of writing recursive pulls scales really really really well so here we have this method that exists in a system of mine called buffer async and the general idea is we want to like pull things as fast as we can and stick them in a bounded queue and then have those elements sitting in the queue so if someone else can pull from that queue as fast as they can and that provides a certain amount of you know parallelism or sorting out of concurrency in the stream processing so that as you know systems are all processing we're pulling more data and so we can define this again recursively in this case we end up using the bounded queue constructor from the async package NFS - and we go and allocate this queue now the return type here is challenging we get this F of Q of F attempt option chunk a the important part here is we've allocated a queue inside of this F type constructor so you can sort of think of the return formatted Q as being like a factory that builds a queue that when we go and evaluate make you that has the effect of generating this queue in memory somewhere and as long as you then use that same queue for all n Q and E queue operations we're safe if you were to evaluate this two three four times you end up getting two three and four queues created on heap right and then with this queue we can do things like let's evaluate it get it into the stream algebra and then flat map that singleton stream that comes back from that and now we have access to the real allocated queue okay and then with that real allocated queue we sort of want to do two things we want to say start pulling from the input and writing the values into the queue and so we use if we chain together a bunch of Combinator's here we say start with some source stream get access to the underlying chunk so inside of FS - all of the individual data elements are actually put in like a collection called chunk it's sort of like vector but it's faster in various ways right for our use cases I can do things like non boxed you know byte arrays things like that so it'll be fast so sort of like get us let us access these internal chunks we have this none terminate thing which wraps every element in a sum until you hit the end of the stream which case it gives you a single none we've this attempt thing that tells us about errors that occur then we write it all to a queue and then we drain the queue we say we don't you know we're gonna write the things to the queue but we don't care about like any output of that the whole point of doing this is just getting them into a queue and so we end up with this is the stream that has no output right it's output type is nothing and then we have another stream we want to create to read from this queue and this is when we want to return two people right so we read from the queue we say qdq and then we sort of undo each of the combi daters we stack together right we say like wreath row any errors that might have occurred and that sort of gets rid of the attempt thing out of our type undone terminate like take all the sums and turn them back into values but as soon as you hit a nun that we're at we're at the end right and then you know kind of undo the chunkiness like take those chunks and stick them back omit them as individual chunks into the stream and so what we get back is this stream of FA and now how do we like you know sort of run these two things and like race them against each other right we want to have someone filling in someone pulling and we use this async operation called merge and in order to use merge in order to do the mate queue we need to find something on F we need to constrain F in some way and so we had this async type class we use some some type class that describes the ability for a type constructor to do things asynchronously so I said we have no dependencies which means we reinvent things and we reinvent them badly because we don't want to provide a third alternative to the functional type classes right we've defined a very thin set of type classes very minimal number of methods minimal syntax just the things we need to get the library built and to start defining you know custom pipes and and and other accommodators and so here you know we have phunkderp liquid monad and Traverse like we don't have a fancy Waals framework we don't have you know all of the you know other types of functors right and we have no plans to write we do not want to be another functional library but we do need a set of data types of a set of extension type classes to do the things we're talking about to remain polymorphic and so we have this catchable type class you can sort of think of catchable as a monad that lets us catch effects that are I'm sorry catch exceptions that occur during evaluation we have a suspender belt i'p class which sorta like monads suspend from think pure script has it you know that lets us talk about like delay right let's just talk about like deferring evaluation of something we have an effect type class which is a catchable suspend a ball that lets us evaluate things asynchronously it's kind of weird and then finally we have this async type class which is an effect that lets us create memory cells and this is kind of cool the idea here is that the primary operation on async is the notion of going and creating a ref and if you create a ref it's like a cell in memory that supports getting and well a supports basically comparing set and for folks that know distributive systems you can build all concurrency primitives with compare and set so we did we have our own semaphore we have our own queue which we saw already we have our own signal we have an entire non-blocking concurrency library like all internally callback based right based off of some execution context or thread pool but built up on top of this notion of an effect full ref an async ref which is very powerful ok so why do we care about all of these type classes right and we've seen this this in a lot of talks this week we care about it because it lets us reason about what's happening we can look at a particular Combinator and get a pretty good clue right away as to what's going to occur inside of the definition that Combinator when we looked at utf-8 to code earlier we saw the effect type was unconstrained so we knew it couldn't have any effects right it couldn't go be going out and reading the network or allocating memory cells or evaluating effects you know asynchronously because we knew nothing about that effect type F and so it's very powerful but admittedly it's maybe a little unsatisfactory when you first come to the library right and just see all these F type constructors everywhere and it's a bit overwhelming so maybe a more practical benefit of this is that there are lots of task types fs2 has a task type there's the scholar Z you know if that task type monix has a task type and there are plenty of other type constructors folks use like Dustin right now is is using the F as free as cats free applied to a specific algebra right and like in my business we have this type called trace tasks where we effectively do like Zipkin like tracing of tasks evaluation and write all of the you know spans that have occur out to some storage system and so we want to be able to you know run streams in that effect type alright so in the last like three minutes I want to talk it look at a system that we built at Comcast it's basically a network analysis tool so think sort of like Wireshark or ethereal like that type of system but run in like a lights-out type of mode so we have all sorts of different capture sources where we pull stuff off the network and we maybe optionally like record the data we're seeing we do some domain-specific decoding of the data we're seeing and then a whole bunch of like rules analysis on different patterns in the data and then write it to a bunch of different outputs and we represented this entire system as a stream and a set of pipes okay and so this particular application as you configure each monitor each thing that we'd go and look at effectively what we would do is build up this big stream pipeline we'd start with this input type that you know read some time-stamped values from from a capture source and then hooked it up to a bunch of pipes that were defined based off of the data types in the system this is like video type stuff right so like if it's like a MPEG video stream we're looking at maybe the source is gonna pull from like a coax device and maybe we're doing some type of MPEG decoding and MPEG rolls analysis on the values and what I want to focus on right now is the capture source because there are a bunch of different capture sources we supported the first was UDP datagrams right like if someone wanted to run the application and they had their own system they wanted to have that system send messages directly to our monitor and let it do its analysis and we could do that because fs2 is got like this UDP package built-in you know built on top of Java and IO you know but you can see like it provides this very high level stream II data you know streaming DSL for working with UDP here we're saying like in the task constructor you know buying to some address we get back a socket if we flat map and on that socket we can like read datagrams out and omit them into our system we did like Lib peek at base monitoring right so we took Lib P cap and put a native wrapper around it and now we've got some resource we need to manage because we got to go to the operating system to the kernel and say like let us you know let us start monitoring the NIC card for packets and then once we start monitoring that NIC card we have a kernel call to pull data from the kernel like pull elements from the kernel we did like RF device type stuff which again is sort of like a combination of the previous two we've got like some like you P&P discovery process we're like when you open a connection to some coax device um some like set-top like device on the network and we gotta can go configure it and go talk to it and get it to start streaming data to us but then we want to start you know receiving UDP and so we can sort of put together like the bracket you know Combinator with like maybe some of the UDP Combinator's to uh to do that a nice way and then we did this um peek at playback where we have like a recorded file from TCP dump we want to play it back at you know maybe different clock speeds I won't have time to go through it in detail but pcap playback is really interesting because you want to play it back at different speeds and we have this state machine to do it and you know I am out of time so I won't show the details but the cool thing is you can represent state machines in fs2 as Co recursive functions using the same exact pool you know recursive pool structure we saw already and you just you know gradually build up these state machines and implement every state as a CO recursive function inside of a big pool so that's all I have I am out of time so a long time for questions but anyone's talk of us to come find me later today