 This session is about reactive things. And I'll introduce myself. I'm Muskaq. I'm from Hot Work Pune office. And I'm a big Scala engineer. I like to speak Scala and talk about Scala. And we are doing Scala projections in Hot Work Pune a lot for the past four years. So I'm planning to evangelize in that future. About reactive things. I mean, what is the familiarity? I mean, have you heard of reactive things and some familiarity, reactive things? OK, so I'll start with the basics. So this work is we are doing a prototype. We are doing a prototype for a customer. It's going to be open source eventually. That's why I can share this. And while doing that, we use our cut streams and stuff like that. At that point, I learned, personally, I learned how interesting it becomes the moment you start using lazy streams or collections. So far, I have been using, I mean, past four years, I have been using Scala collections. How many of you are familiar with Scala? And have you Scala, to some extent, just some try? So Scala, if you Scala use Scala collections like every other minute, you are writing something using a Scala collection. And that model is very simple, but very powerful. So what you do is, for example, there is a disk or a source somewhere with data. The first linear graph says that if you want to just process that data, what do you do? You first rate it. And then the three cells, they represent that, OK, data has come in memory. So it's as simple as that and as dumb as that. So it's in memory collections. And then you do further processing like square, all the numbers, map, dot map, square. And you get yet another collection in memory. So that is how it works. And then finally, you say take two. And then you get the first two elements. Again, yet another collection in memory. And then these collections are immutable, which is good, because then they can be freely shared. And we'll see that in the next graph, how sharing is easy and possible. But in memory collections have a limitation, right? I think if my file is any decent file of size, say, 1GB, I can't do this kind of a processing. So I have to use a different way of processing, not the Scala collections in memory collections. They also, at each point, they also allocate a new set of collection after each processing. So if you have many steps, then there is a lot of garbage which is produced. Another argument is that maybe because you're producing, you have to iterate over each element. So for example, when you say square, at that moment itself, you have to go over all the elements and produce a new collection. And then again, iterate to do the next processing and all that. So there are some downsides. Garbage in multiple iterations. And in memory, so I can't use it for IOProgramming realistically, right? That is the good side is that if you can use Scala collections, you can use them without the immutable collections, without thinking too much. For example, I'm calling this a graph. But you create these graphs every now and then. You don't even think about them. You read in memory. And then you share this in-memory data structure of 1, 2, 3 with two functions. One function iterates, does a map, takes something, produces a collection. The second function does a square and takes something and produces a collection. Now it's a proper graph, right? I think you are reading from the data source once. And then you are sharing that in-memory copy with multiple branches of your computation. But you don't have to think about it. If you're done programming, you wouldn't say, oh, I created a graph. I mean, you just share, you share, right? Because immutable, it's in-memory. So that is the beauty of in-memory collections. That's why that is the bread and butter and that is the workhorse. And I will use that all the time. But not suited for IOProgramming. So before we go, just let's see what happens. Can you see the fonts? Or should I increase them? Increase? Sorry? Fine. So I have a list of numbers, right? 1 to 20. And what I want to do is I want to make them into in-memory collection. So I say two lists, which is in-memory. And then I want to read them. So reading, they are already in-memory, right? So what do I read? Just to log, just to show you the effect. Read is just for side effects. So if you see the definition of read, square, double, they do the operation, read doesn't do it. Just returns the value. But it logs so that you will know when I run the test. Same for squares, square doubles, but also logs. And now I fork to create a graph. I fork it. And then what does fork do? Well, it takes that excess. And then on one branch, it does the square. So it takes that immutable collection. Does a map, square, take five. And then for each ignore. I have logged, so I don't have to do anything for it. And then in the second branch, it takes the same source in memory and does completely different computation. Because these are immutable, they always produce a new copy. You can freely share. And this is the example of doing that. Now it works well, but it works in a way that take five and take ten doesn't have any effect. Well, it has an effect in the final produced value, which I'm ignoring. I'm just printing line. But on the side effects, how much competition is required, it doesn't have any effect. There is no short circuiting. How do you see that? Let's run this test. So what you see here is, well, it first goes and ruins all the 20, because they're in memory. So it has to basically, when you say map, square, it doesn't know that eventually you're going to take only two or only five. It has to go do the job. Then it does the same for squaring. It reads all, it squares all, and it doubles all. And then in your program, you actually wanted to just take five from one and take ten from one. But that short circuiting will not happen. So this is a simple and nice example, which shows the limitation of in-memory collection. But you get the picture, how the graphs are produced. OK, so I'll move on. But are you on the same page now, where we are going? OK, cool. So let's start with the lazy pipelines. So lazy pipelines are not default in Scala. They are available, but they are not the best practice. You don't default to them, but in other languages you do. So what it means is, let's understand the simplest version of a lazy stream. It's a pull-based stream. You have the same source, and then you have a pipeline. You want to do read, square, take, and sync. I'll talk about that. Now the difference between the earlier version and this version is that I'm not doing in-memory processing. I'm not getting loading the whole data in memory and then processing the whole data into another memory collection. I'm not doing that. I'm doing it in a pipelining fashion. That is the word, pipelining. That you kind of stream one item at a time. But how does streaming trigger? What triggers the streaming? And how does which direction does it move? Those are the arrows, actually. So sync is a forage print line statement. Assume that that is a sync. Sync is a terminal operation, which does some side effect. Now that terminal operation, so you see that our operation will look like this, source.read.map square, and this is the terminal operation. Now if this is a lazy pipeline, this terminal operation will call the next on the earlier stage. So each of the stages have an internal iterator. And iterator API is very similar to the Java iterator API, which has a next and a society technique next and has next. Now each right side will call the next of the iterator on the upstream. And once that call is obtained by the stage take, take will call the next of the iterator of the earlier stage and so on. And then the call chain will happen and go all the way till the cursor of the database or the file. And then you will get the first element, which is one. And then it will be processed here, logging. It will be sent to this, squaring, remains the same. Take two doesn't have any effect because it is less than two. And then it is given to the same. So you know how many hops are going on, right? Well, it is not coming in memory. It is pipelined, which is great. But then there are many hops just to make imagine if these are network boundaries. If these are network boundaries, then for each element to stream, your next call will make across the network to the previous stage. And then across the network and across the network. So one, two, three, four, and eight hops. And you'll get one element. Maybe you have better capacity at each node to process faster, but you can't because your bottleneck is now the IO. And that is how this paradigm is, even though it's better than the in memory, it's not ideal for the IO program. You understand this part now, how it's happening. But there is an advantage to this part, which is there is a short circuiting happening, right? So for example, I have three elements. I could have three million elements. But because I'm saying take two, only two elements will be streamed. So there will be eight hops. And then eight hops, and I get two elements, and I'm done. So I will not be processing the remaining of the three million, which is great, which comes with the laziness. The trigger is always on the right-hand side. The demand is always on the right-hand side. And then the demand basically goes hop, hop by hop across the boundaries of stages. And then data comes from left to right. So demand goes right to left. Data comes from left to right. Is that clear? It will be same. Even if it goes that way. Yes, it will be same. OK, so now if you're not familiar with Scala, you can compare. I'll give some example. This is similar to a stream in Java 8. Are you familiar with Java 8 streams? How many of you? Most of you, right? So Java 8 streams will behave. I mean, of course, they have other capabilities. But if you just see their basic capability, this is what they will do. So this is about the pull model. Well, let's talk about the push model. Because we already explained some of the limitations of the pull model, so many hops, right? So many hops, and waiting for the elements to arrive. What is the alternative? If you don't want to do this way, and you still want to do efficient IO programming, what is the other way, the push model, right? What does that mean? Oh, source on the left starts pushing on the right. After appropriate trigger, we'll talk about that. But it just starts pushing on the right. And then right side, so for example, if there is a source 1, 2, 3, first, that one will be pushed to this stage. That stage will log it, and it will push immediately to the next stage, and next stage. So there is no request. There is no next call. It is just being pushed, right, without you. Maybe if someone triggers it somehow, and then just data starts flowing from left to right. There is no continuous next, next calls happening across the boundary. So across the boundary, that is the difference between the pull and the push model. Now, we have to talk more about these arrows. So these arrows, they represent the asynchronous boundary, right? Which means that because now you are in a push model, you can afford to be asynchronous. What does that mean? Well, each stage will have a place called buffer, where you will store. As upstream is given me some data, so we'll put it in the buffer. And then upstream is free to push next data. So your buffer will keep growing. But at the same time, on other event loop, you will be pushing. You will be logging your data and pushing it to the next stage. So your buffer will be draining also. So buffer drains on the right side, and fills from the left side, right? So each stage is symmetrical in that sense. Every stage has to have a buffer. And that buffer gives you the power of throughput, high throughput. Now there is no waiting of the next call and all that. Just I'm sending the data. I'm sending the data. Well, some sources don't need any trigger to do this. They are always in this mode, in a conceptual, in a domain, in certain domain. For example, if you want to model the time ticks in a reactive world, and if you model it like a source, well, that source is not going to wait for any trigger. It's going to tick. Basically, if it's a tick every second, it will give you an event, whether you use it or not. Similarly, mouse, if you want to mouse click, mouse scroll, mouse any UI event, if you want to capture as a stream of events, well, that source doesn't need any trigger. It doesn't wait for a user basically to use those. It's just producing a series of clicks. Someone maybe uses it or just discard it. Those kind of sources will summarize them in the slides, but those kinds of sources are called hot sources. And that is the variation of the push-based systems where the sources don't need any trigger. They just, they are always pushing the data whether you want to use it or not. But other sources, which are very relevant for IO programming, for example, if I have a web service, which is giving me data, and then one web service, read processing it and processing it, well, I need a trigger. I can't say that, okay, there is a web service which is continuously giving data. Well, it can stop. I mean, it is reading from the database, which is not dying, right? So it can wait till someone wants it. And that is the trigger of a sync. So the sync on the right-hand side, that is the trigger in the most push-based systems when the right-most terminal actions. So if you know Java, it streams, there is a special kind of actions called terminal actions, is that right? Right? So those are the terminal actions which are syncs. They give a trigger. Unless you attach the terminal action, data flow will not start, is that correct? In Java, it streams. That is what will happen even in the push-based systems which wait for a trigger. And those systems are more important for IO programming. And those systems are called cold sources, right? So that is just a terminology. If you look into reactive streams, you will come across these two terms. Okay. The third thing is, you see, on Next. Now, instead of Next called going here, there is only one call going in that direction. Now, if I have to send data, if I have to push data, I can't call Next because Next comes to you, right? You have to push data, so you have to invert. You have to have a dual of Next, which is on Next. Next is a call which gives you T back. On Next is an event in which you put T, basically, right? And then call it, you call handle of the downstream so that it's pushed. So you keep calling onNext, onNext, onNext, and buffer keeps filling. And that guy processes and keeps onNext of the Next. That's why, that's why not Next, onNext, right? And that's why it takes the element rather than returning an element. So you appreciate the duality here. So asynchronous boundaries need for a trigger or not, and the events which are passed using onNext. These are the three important things. Do you see any downsides? I mean, this is like a very interesting powerful model. It will obviously very high throughput because you are avoiding so many hops. If sync wants to process Next, it's already available, right? Most likely, the previous stage will have it. So it will be definitely much more high throughput as compared to the pull-west model. But do you see the downside? It's too much of a buffer. Right, right. So right now, if you really implement, if you really implement this kind of a system, which many people don't end up doing that because there are, I'll come to that, then you have an issue of buffer overflow because you can't determine the rate at which the network is operating, the stage is operating. So upstream could be pushing at much higher rate than the downstream is draining. And these streaming systems are not ephemeral. They run, they run for days sometimes, right? Or always on. Sorry, I think I should just activate my caffeine here. Right, right. So I think this doesn't work. It has to just come out and then go in, and then I'll wait. So trigger, there is a trigger, but it's a single trigger. It's not that for each event you need a trigger. For the pipeline to start flowing data, you need one trigger, and that's it. Yes, for the next event, for each event you don't need a trigger. You just need to say that, okay, this cold pipeline has a blueprint and then it needs to be started, data needs to flow. And you just attach a sync and that trigger saves. I don't know why it takes so much time. Oh, it depends on your streaming. For example, if you want to have this streaming service, it will run forever. Ideally, it should run forever. I mean, there are. There's a lot of services. Right, sure, yes. But if that web service is like a, say, a ticker data for your share prices, then it will keep going. It will keep going. There is like no end to that. Theoretically, there is no end to that, right? Yes, in streaming systems, people will expect that it is in order. It is in order. So even though I'm pushing on the buffer, right? Buffer is ordered and I will, yes. If you're doing take two, the sync, attaching the sync is a conceptual operation called subscription, right? Someone has, terminal action has subscribed to the blueprint, right? If you say take two, there is an implicit unsubscribe, which is the final trigger. For example, if you don't want this operation to go for infinitely, what you will do, you will say that unsubscribe. You will call it or you will say take two. And if you take any reactive library, take two internally goes and unsubscribes. The moment it unsubscribes from, it will unsubscribe from take, right? Oh, sorry, take will unsubscribe from square. And internally, square will unsubscribe from read and then so on. Because the moment you attach a sync, it's not a magical operation. The signal has to transfer. So the attaching a sync is a conceptual operation of subscription to the previous stage, which triggers the subscription of its previous stage and so on. Right? Once, to trigger. Subscription, yes. You have to attach and then subscribe, subscribe, subscribe. And then it starts flowing one after the other for a year, right? And then you say, I'm done. So you say unsubscribe, which is implicit in take two. And then it unsubscribe, unsubscribe, unsubscribe and then it removes. So it knows, the source knows that I have just zero or fewer subscribers now. Where to push, right? So that is the implicit thing, right? Well, sync right now, I have just, it's a conceptual. I mean, depends on which library we are using. We are not going into details of one implementation right now. It's a conceptual that there is a sync represent a subscription. And it will subscribe whenever you attach a for each print line. It's a sync for us conceptually. It will subscribe to take, which will subscribe to and that will be the take. Right, I think we are not, I'm not going to deal with that because I haven't figured it out in the, how to cover that here. So we are not dealing with fault tolerance. So if I say always on stream, what happens when the node goes down, right? Fault tolerance of stream, it's a big topic. So I'm not going to cover that. But yeah, I think it's a concern. Okay, so that is about the push model. Is it clear now? So pull push is like, and so roughly, I think in my definition, you will see reactive stream jargon at a lot of places. Roughly, if you have a lazy pipeline and a push based, which is the second element, you are talking of reactive streams, roughly. But you know that there are big problems there, buffer overflow, it's not practical. I mean, if someone says, this is the model, I will not use it. So this has to be fixed. Well, how is it fixed? There are two ways, right? One is cheating actually, which is when you attach a sink, right? You attach a sink, I say for each, print line. And suppose the print line is a very slow operation, right? It takes five seconds to print. Well, that is your back pressure. So what happens that your sink doesn't release, so upstream take is going to call on next, on sync, right? And you say that I will not, and on next is supposed to return quickly, right? When you push, our next call is supposed to just buffer it and return immediately, right? But my sync says that, no, I'm printing, you got me the data, I'll print it. And unless I'm printing, I'm not returning the call, even though it's a wide call, I'm not returning the call. So because of that, the upstream will not be able to push the second element. So it's a natural back pressure, but that's really cheating, because now the throughput will again go down to one element. So effectively, you will have, say, a buffer. So the back pressure, and then because, because there is, I can't push next, right? What take will do, take will do, is that it will block the on next upstream. It should not, but the implementations do that. I mean, many implementations like RxJava, if you see traditionally, you get back pressure free. You solve the buffer overflow problem free, because you say your terminal action is blocking. I'm rendering on the UI. I got a stream and I'm rendering on the UI, so unless my rendering happens, I'm not going to give you on next back, and that's why you won't be able to push. And the previous stage will do the same to the previous stage. So you have back to the back pressure. So in that sense, some advantages of the push-based systems will go away, but that's a very common thing, and that's why you won't see this memory issue so often if you use those libraries. You get that, right? But if you want in a truly push-based system, with a high throughput where you want buffering, and you want to, the principle is that on next should written like immediately. You should not do anything in the event loop, which is for each stage where we are calling on next, then you have to do something about it. You appreciate that, right? Okay, so the second approach to solve this problem is a recent approach where people have said that, okay, let's standardize a way in which we can control this back pressure, and that is, I call the hybrid approach. Well, it's easy to explain, so it's kind of a combination of the two. So it mainly remains a push-based system, but there's one more contract which is followed in the other direction. And that is similar to the next call. You see the next call on the top? Next call always demands for one element, right? Next, in a iterator. Here, your next call can demand for more than, I mean, depending on your setup and your library and your infrastructure, you will, you can request more, more than one. So what happens is, so instead of next, they call that call as a request. It's just a terminology, but it's just like a next four, basically, right? Important difference between that next and this request is that the arrow, right? Here is a dotted arrow. What does that mean? It's a asynchronous call. It's like a fire and forget. It's also fire and forget, right? So I say, my upstream that, okay, give me 10 more, right? Because I know that I'm going fast, so my buffer is going to be empty soon, so I'll say 10 more. And then this request when reaches take stage, it says, oh, this guy wants 10, I don't have anything left, so let me demand 20 more upstream because I'm processing even faster, right? Something like that. And then this demand will continue to go. And it's not a one-time call, but it is not a call for each iteration also, right? So this fact that you can make this call in number of times, this call you will make asynchronously so there is no blocking across the network, right? And the upstream will, will honor the request that, okay, if downstream has demanded 10 now and 10 previous second, and I haven't sent any yet, okay, I can send 20 because there is a total demand of 20, right? 10 previously, 10 now. So it can batch. That is all implementation detail. It can batch. It can say that now I have a demand of 20, let me efficiently send that 20 to that. But it will not send the 20 first. That is guaranteed by the contract. And that is nice because the downstream knows how fast and how quick the processing will happen. And that demand is conveyed to the upstream and it is batched. And if you honor that, okay, you will not send more than the demand, then the buffering problem and the overflow problem will be taken care of. We appreciate that, right? Okay, so that's a recent initiative that is the many, many companies are involved in this. Scala communities involved in a big way. And that is called as reactive streams. Protocol, I mean, sorry. Right, which means that each stage will have buffer. So it's not that, okay, I don't need any memory, right? It's like in the first stage, if you see pull-based systems where you are making one call, you don't need anything in memory. Maybe one element is flowing in transit. Here, each stage will be buffering 16, depends on the implementation, four, 16, 64 elements, right? So there will be some memory, but it will be, it will be predictable. Once you know, once you have configured that, you guarantee that that memory will never exceed when you deploy it in production, right? But because it's buffered, you get the high throughput. Because now when I have to send data to the next guy, that data is right available in my buffer, I don't have to make all the next, next calls and then get data and then send it back. So you combine throughput with safety. It will, it will take time, but there is no waiting. For example, the demand is asynchronous, right? Which means that the upstream say, oh, I need 10 more, okay, let me request 20 more. And the upstream is also empty. So it will keep going and maybe there is no data flowing downstream, it's only the demand is accumulated and the demand will stop. Because downstream will say that, okay, I haven't got any, any elements. So I, so the algorithm will not ask for more, right? It will, it will just wait for next 40 because my buffer size is not more than 40. So I can demand 80, right? So it will just not demand and it will just stall. But that stalling is different than the, the waiting and the blocking, blocking style. So, so you haven't, you haven't hold on any resources, any threads or any network calls. It's like suspended. The whole thing is suspended. All resources are free. They can, they can do whatever they want. But the stream is live. The moment there is a, there is a drop of data on the left-hand side, it will start flowing. Number of calls, minimum number of calls for every data to happen is the number of network boundaries, but in one direction, right? And that too, not waiting. For example, when I need the next item in the, in the, in the pull-based systems, how many calls I do? I say one, two, three, four, five, six, seven, eight, right? All that, that will be avoided. But that doesn't mean that I will still have four hops. No, because the data will always be there in the buffer. Always be there in the buffer. So, so the moment I need one, it will, it will come. If my buffer is empty, well, there. Then yes, then it will be at least, if my buffer is absolutely empty, right? My source is really slow. Then it will become like a, it will, it will, it will just work just, just in time. It will just, there will be one element in the in-flight all, all the time. But even then, there will be only four hops. It will not be eight hops, right? So, it is still efficient. And it's principally sound for, for high throughput, right? Okay, so, yeah. It depends on the implementation. So, implementation can be smart and say that, okay, I instead of, so, so, so demand person can say that, okay, I will not request one, one, one at a time. I will request 10, 10. And then the supplier will say that, based on our, our implementation, I will not supply 10 each time. I will batch till 40 and then supply together because that is more efficient. Yes, go ahead. It depends, it depends. So, it can, or, or it, yes. In my example, it will just one call, one push call. But it could be the other way round. The demand is happening at a gross number, 50. But you know what, I know that I, I never get 50, right? So, I will maybe, I will push all the data that I have. Or I will say that because of some algorithm, I will just push 10, 10, five times. I mean, that is possible. That is, that is left to the implementation. But there's a possibility that data is just close to you, right, if it is buffered properly. And it is safe. It is not going to overflow. So, that is a safe. Okay. So, if you, if you formalize this, I mean, of course there are a lot of corner cases. If you formalize this, then what you get is the reactive streams protocol, which these, these people are promoting. So, you, if you just search for reactive streams, you will go to the GitHub page. It's a, it's a detailed specification. And there is a test kit. So, which means that other people can implement it. It's a way of thinking like, it's a JDBC, right? JDBC is a, is a protocol or a standard. Now, many other people, drivers will, will implement that database drivers. Similarly, this is on the similar lines. For example, now MongoDB, right? MongoDB has a driver. MongoDB has many drivers on the JBM, right? For Scala and for Java and for RX and all that. One of the drivers for MongoDB is reactive streams protocol, right? So, they say that, okay, this is a driver which follows this protocol. See, Cassandra will do that. And many other data sources will do that. What, so, so what will happen? Because it's a, it's a standard. The, the APIs are defined. There are four interfaces and, and nine calls or something like that. And all of them will follow. They, they would have passed the test kit of the, of the specification. Libraries can interoperate. It's, so this protocol is very minimal. It just says that how to, how to control the flow. It, it doesn't give you all the jazzy things which RX Java will give you, for example. So map, flat map, filter and all that. It, it doesn't care about that. It says just, okay, flow control. What, what, what are the primitives there? So now what you can do is, you can, if RX Java, now which supports back pressure to some extent, and say, Accastreams which support back pressure, and say MongoDB driver, which just gives you bare bones reactive stream. You can use that driver to make a MongoDB call. And you have a handle to a publisher, publisher of documents. And now you publisher, you wrap in Accastream and use all the jazzy API of Accastream or wrap it in a RX Java and then use, use. And end to end back pressure will be guaranteed. Because these libraries, RX Java and Accastream will also implement the same protocol. So when you are doing the integration across systems and they are using different technologies, different libraries which is very likely, if they follow reactive streams which is this hybrid protocol, then high throughput with the back pressure end to end will be guaranteed. That's, that's the hope. Okay, so let's, let's move on. I think this is just like a summary. Pull versus push. Pull-based is synchronous, it's blocking. And there is a natural back pressure because unless I process my element, I don't ask for the next. You can, you can do, you can be innovative and then do concurrently in threads, but that is still blocking, right? Push-based is asynchronous, it is non-blocking, it's high throughput, but slow downstream will cause a buffer overflow. But we, in the fine print, we say then blocking subscribers. If all the subscribers are blocking, then you get a back pressure. But you also lose a lot of benefits of the high throughput. And then the hybrid model is the best of both worlds in a way, so that's the summary. Okay, this also we saw hot and cold. So hot is where data flows without a trigger and the cold is, you need explicit trigger. And the examples of the cold are IO centric things like web services and file system reads and all like that. For hot it is mainly UI events and time and Twitter feed and stuff like that, right? So those are the examples. Is that clear? No, no, we cannot, because hot by nature is a hot source. There's no notion of back pressure, forget it doesn't even care if you exist or not. It just, it just so Twitter stream, right? It's like, yeah, it's dropping the buffer. It has a streaming API to which you connect and then show it, but yeah, if you don't connect it's like buffer is dropping and then whenever you connect tomorrow you'll get from that point onwards. So the, it's a good question. So the concept of back pressure applies only to the cold sources and that's why it's more relevant for IO programming and not for the UI programming because UI programming, yeah, there are so many, I do a mouse drive, there's so many events. Yeah, if someone is not handling that they will be dropped. I mean, I will get the event where I connect it. So I think I will summarize that in one of my slides. Okay, I think good. So, but before I do the demo next, I need to talk about the graphs now, right? So we looked at this graph, right? Initially it was so simple within memory collections like it was so predictable. Now you have to think a lot. So let's look at a graph, the similar graph that we did for in memory collection. If you do it here, now we had two possibilities suddenly, right? One possibility is you make two branches from the start to the beginning. You read from the source, you square, you take two and then you print. And the other thing also starts with the source but doubles it and then takes it and all that. So it's actually two branches right from the source, correct? Versus, you can do a second approach where you say that okay, you read. You read data once from the source because it's a network call and it's expensive so you want to do it once. And then you want to branch after that. You want to say that okay, after that, like in memory collection, you basically say that I want to somehow share this data so that my network calls are not repeated because those are expensive, that's a very valid concern. So you will have some junction basically and that junction, after the junction you will do map, square and all that. The moment you do that, the first is clean. This is clean. This is as simple as say in memory collection. In memory collection, you don't have to think. Everything comes in memory. You can share immutable and goes on. Here also everything starts from the beginning. Everything starts from the beginning. Oh, you don't have to think. Nothing will be in memory and all that. Everything starts beginning to end. But the moment you want to optimize, now you have to think. Most likely you would want to do this way because these calls, you don't want to make, all the subscribers make a DV call right away. I mean, you will have to, if you know that there are concurrent subscribers, they should get the benefit of reading it once from the database, right, for efficiency. How do you do that? How can I ensure that this works? Well, it depends. It depends whether you are using a lazy stream. Of course, this is a lazy stream. It depends whether you are using push-based stream or a pull-based stream. And what are the options? One option, any gut feeling, what are the options? I mean, it's written there also. If it is like a Java stream and you want to do this, I mean, Java stream, assume that they allow you to share. They allow you to share, basically, somehow. What will you do? No, it's a pull-based system, for example. Iterators. And you want to share. You have read from the details. What will you do? Copy. Copy will be required, right? Story, in memory. You have to convert your nice not-in-memory thinking into in-memory thinking. You have to buffer it somewhere, right? You have to say that, okay, you know what? How many of you are familiar with .NET collections? Few, few, right? How many of you are familiar with Spark programming model? Some of you, right? Okay, they are by default. If you just program the first way, they behave like that. So in .NET collections, they are lazy. They are not like scalar collections. So each time when you apply, they don't do in-memory processing. Only after the terminal action is applied, you do all the passing. But they then fall in this category of the, for example, they will be reading from the source twice. While in-memory collection would have read it only once. So there is a downside. So lazy collections by design or by default are not superior to in-memory collection. They have different use cases, right? So I think that is a very important point because people are very excited about anything lazy. But that need not be the case. So what do you do? In C-Sharp now or in Spark, you want to avoid duplicate reading. So what do you do? Any C-Sharp guy can you tell me? So you want to share. You have read it from the database in a numberable. And now you want to share that information to two downstream functions. One is going to square it, one is going to double it. What will you do? Spark IDDs, I'm talking of IDDs, right? Spark IDD or C-Sharp I-Numerable, what will you do? Can you just share the I-Numerable like that? Well, you can. But if you share just I-Numerable, then it is still in the first category. If you want to share, what will you do? You will call tool list on it, is that right? I mean, I haven't told, I don't use the .NET, but when I asked, you will call tool list or something like that and convert I-Numerable into a in-memory collection, even in .NET, and then share it. That is how they do it, right? In Spark, IDD, what do you do? You say .cash, you say .cash at that junction and then information at that junction will be cached. Well, how to cache, this is called memoization. How to memoize, there are many different possibilities, many possible, you can be very smart there. We'll see very, very briefly. But you'll have to memoize. So you'll have to get some data in memory and you'll have to keep that data in memory till the last subscriber has seen it. And if you don't know when the last subscriber is going to come, you have to say that, okay, how long are you going to wait? But that is not a good criteria. How big you can grow? And after that, you start dropping. So the subscribers which come later, they will see the data from that point onwards, but other subscribers will see all the data, right? But you appreciate the moment you are in a lazy stream world, you have to think of junctions and what kind of a memoization is happening. And caching in RDD is very sensitive topic. You have to really know when to cache. If you cache it the wrong place, I mean the whole world will be in memory, you'll, the cluster will come down, right? So it's not as simple as it sounds. Okay, sorry, just a second. Yes, yes? Yes, yes, yes, yes, yes. True, very true. We are, we are. If we, if our memoizing is not smart, right? Then we are. If our memoizing is smart, which means that, you know what? One subscriber is fast, one subscriber is slow. And the difference between them is like growing, but maybe it's small of 10 elements, right? And it's growing, it's going to go till 1,000, right? So I'll have to buffer only 1,000 elements in memory. Is that right? Because the first subscriber would have seen all the elements till that point. Second has seen all but these thousands. So I will have to buffer. So if your buffering strategy is smart, you're still better off. If your buffering strategy is like, say, cash in RDD or C-sharp tool list, then maybe it's not very smart because you better filter a lot of data till that point because everything is going to come in memory, right? You see that. So that's the difference of how you memoize. So you're still better off. And if you're in a, this is still pull best world. So let's see a demo. Let's see a demo of this first. So I'm going to go to the pull best. I'm going to deal with the numbers that we saw, one to 20, right? But I'm going to first convert them into iterator just to show you a streaming. Iterator is similar to Java, it streams. And it is a pull best but a streaming system, a lazy system, right? And then I will do map to read. And then I will share it in a fork iterator. Now what does fork iterator do? Oh, well it's very similar to what we did with the eager collections, right? It will have two branches, one for squaring, one for mapping, and it will take five in the first. Okay, any guesses what will happen? Iterator, like Java iterators or Java streams are perishable. What does that mean? I mean, as you use the data, that data goes away. I mean, that is not, right, it's perishable. So which means that I have two subscribers. So first we'll consume some part of it, and the second we'll get data only from that point onwards. So far the data would have been consumed, consumable or perishable, right? So let's see if that is the case. Well, what operations I'm doing? I'm doing a take operation for five elements, and after that I'm doing a take operation for 10 elements, but the first is square and second is double. So let's run this. Okay, so let's see what happens. Well, I got from one to, I could square one to five, because that's what I wanted to do, and it rate from one to five, and then did the square. But then for doubling, it started doubling from six to 15, those are the 10 elements. I did not get one to 10, I got six to 15, right? Why? Because the stream was perishable, and the data got, if you do it in Java, maybe there will be an exception because the iterator may not allow you, stream may not allow you to attach multiple subscript, but that's just the implementation detail. So let's go back and try to fix that by doing some memoization. I'll explain the strategies later on, but just see some memoization. Well, if I do some memoization, what I see is what I want. So I see one to five being squared, they're red, but I also see that doubling is happening, doubling is happening for one to 10 and not from six to 16, but there is another interesting thing happening in doubling. Can you notice that? There is no reading only for one to five, right? Because that is memoized, that is memoized, and that depends on how I have memoized, but that is memoized. So any perishable stream can, you can convert by memoization into a reusable stream, but some data will be in memory, right? It will not be the whole data, but some data will be in memory. There are different ways of memoizing. I think I will skip that for a moment and just go to the reusable streams. What is the example of reusable streams? I mean, I already talked about perishable was javide streams and iterators. Reusable is, that is persistent, that is, I'm talking of RDD, Spark RDD, or C-Sharp Ionomerable, right? You can reuse from the beginning. And the Scala equivalent of that is called as views, that they're very similar to RDDs or your Ionomerable, and on the numbers I am first converting them into a view and then doing a fork, right? So let's run this. So this is simple, right? This is, I haven't memoized anything. I'm just running a view, I'm getting a Scala view, just like an RDD, and running it, just like Ionomerable and running it. What do you see? You see very similar to what you saw last time, but with one difference, right? Reading is happening again, which means that the stream has started from the beginning, right? Because it's a, there's no memoization, but it's not a perishable stream like RDD, so it will read from the beginning, right? Which may or may not be good, depending on what you're trying to do. So, but now if you want to avoid that, what you will do? You will still memoize it, even though this is reusable, you will memoize not because it's perishable, but because you want efficiency of avoiding reading twice, right? And then you will be back to the same effect. You will see one to 10, but the six are not read, one to five are not read again, right? So, this is the effect of memoization. And before I tell you the strategies here, these are some of the summaries, basically, right? So memoization is for pull graphs, it's risky. We should know that it's coming, something is coming in memory, so you're right, it's not as simple. And there are multiple strategies possible. One is the simplest is the full. You say on a Scala view, if I call tool list, it will come all in memory, right? It's not only the five elements, it's everything will come in memory, right? So that is one approach. The same for tool list or RDD cache. There is another way called incremental, where I say that two stream, there is another data structure in Scala called streams. It's don't confuse it with reactive stream, it's like a collection stream. It's very similar to closure collections. Like views are very similar to say, .NET collections. The stream in Scala is very similar to closure collections because if, how many of you are familiar with closure collections? Some of you. What is the behavior? Behavior is like, it is lazy. It streams the data, right? But it memoizes. So next time you make the same call with the same argument, it gives you a memoized value and it doesn't compute the stream again. That is the semantic, which is the exact semantics of streams in Scala. So that is a smart way of doing an incremental memoization because if I memoize using stream, and later on I say take two, only two items will be memoized, right? Versus tool list, everything is memoized, but then I take two, right? So this is slightly smart, but it is risky because if you don't take two, then it is as good as memoizing everything in memory. The other thing is, other version is directly using streams, not using views at all, which is directly using closure collections for these kind of pipelines. That will also give you the benefit of memoization, but it is automatic, which means that there is no step where you will say, okay, at this step, like RDD, like Cache, or I2Stream, or I2List, it will be automatic, which is very dangerous because you will not keep track of where all the memory is being, can potentially leak, right? So not preferable, but you can do that programming in Scala using streams. Closure sequences, which is closure collections, by default, behave like that. That's why they have some power of lazy computation. Well, there is a minimal version, which is possible that I talked about. If you're using Scala iterators, there is a method called duplicate. That duplicate, one iterator will be split into two iterators, and if, but they will point to the same source, right? Same, for example, if there was a in-memory collection, iterators drain the collection, right? Iterators basically will consume. If I get a iterator handle, it will get consumed. But now if I split it, and there's another user of the second iterator, it's slower, the duplicate will keep in memory, only the buffered, only the difference between the speed of the first consumer and the second consumer, the way I talked about efficient. But that is the implementation detail. Another summary that a lazy pipeline, a lazy pipeline can be a pull-based or push-based, that I will see later. Pull-based pipelines are driven by iterators. Terminal actions trigger the data flow, right? They could be reusable or they could be perishable, and we saw the examples of both reusable and perishable. Both of them, when I say reusable, by default they're reusable with multiple consumptions, when reusable means they will always begin from the top. But if you want to basically short-circuit that and have junctions, then you'll have to cache or memoize and then reuse is possible within a signal consumption. Is that clear? Does it put everything in perspective for pull-based systems? We'll cover the push-based system. Just a quick time check. We are, when did we start? 10.30, right? So it's like almost one hour, what half an hour to go. You want a break or is it? Sorry? It's fine. Okay, cool. Okay. So let's look at the push-based systems now. Similar to perishable and reusable, the terminology here is hot and cold. Hot is a kind of a perishable, right? It's perishing. I mean, whether you use it or not, it's like it's going away on the time axis, right? Similar to memoization, the terminology in the push world is called as Unicast versus Multicast. Unicast will just have one subscriber basically, right? Multicast will do some magic and will allow multiple subscribers and will push the junction, push on the junction. So it's similar to the diagram that we saw here. The same diagram, but in the context of now push, we are talking about, right? So let's see what is the effect. So if I have a hot source, like at mouse events and stuff like that, how I'm creating a hot source, I have a number stream. Well, what is the number stream? Number stream is like infinite number starting with one. So I just get an iterator from one. Iterate from one will give me infinite stream of ones. I throttle because if I just use these as a basis of a hot stream, it will be too fast to demo. So I'm throttling it every 50 milliseconds, I will emit the next number, right? And then I'm just doing a map so that I log and I see that every 50 milliseconds the read is happening. And now this number stream I want to use as a hot. So hot is my work around. So I'm using a Custreams which by default do not support hot streams. So there's a small work around basically to make them hot, right? Behave like hot. And then I don't multicast it. I say that okay, only one subscriber is allowed. But I try to attach two subscribers. So the two subscribers are interesting. One is a square squaring flow and one is a doubling flow, right? And then the square flow, because this is a push-based system, I'm throttling them differently. The source is throttled at 50 milliseconds. The squaring flow is throttled at 200 milliseconds, which means that every fourth event will be, this will get. And the doubling flow is throttled at 100 milliseconds. Every second event it will get, right? So I'm trying to attach both. I'm trying to attach, take that initial source and attach two branches, basically see what happens. Well, it will be an error because this is a unicast. So I haven't done the multicast yet. So if I run this, well, error is fine, but you appreciate that because it's a hot stream, it's not waiting for the subscriber to attach, right? It started doing its job, which is reading, right? And logging. And you see that number one to four are being logged, right? And then the first one is getting squared because that reached. But then the, I'll just take this, if it's hot, it'll just... After first, the squaring has happened with five, right? I mean, I'm just printing the number which is being squared, not the actual value, right? So which means some elements are being dropped because it's a hot stream, right? I mean, the elements will be dropped, right? But the moment I try to attach the second after a gap, I'm attaching it after a gap if you see here, so in the fork. I sleep for 500 milliseconds and then I attach the second floor, right? So because of that, you will see that there is an error because this is a unicast. Now let's do a slight modification and make it a multicast. You can make it, a custom will allow you to basically multicast a source, and if you do that, then you will see a similar behavior but without an exception, okay? So let's see. It's still a hot stream, lot of elements are being dropped, but squaring is now happening for one, it happened for two, it happened for four, and then eight, right? And doubling also started after 500 milliseconds I attached the doubling part, so doubling started happening this point, right? So this is the first number which got doubled, eight, correct? And then the second number was 10, and then 13, right? So both are working, some of them are getting more elements because they are throttled less and some are getting less elements because throttled more, right? So you see the behavior of hot and multicast. Now what happens if it's a cold stream? It's a cold stream, I don't have to do anything, it's my stream is by default cold, in our custom everything is by default cold. If I do the same thing with a unicast, what will happen? Let me run it. Okay, because this is a cold stream, and I'm doing a unicast, it is similar to I-numerable, RDD, it will always begin from the top, but this is not a pull model, this is a push model, right? So you read one, two, you have managed to do square one, two, you read, you manage to do all the squareings. So squaring happens with three and the squaring happens with four and five and so on. So first 10 numbers will be squared, right? So you get all the 10 numbers squared and you get the first five numbers doubled, right? So you have this doubling for three and the final doubling is happening for five, right? So you get everything, but you read twice. You read one and you also read one. You read two and you also read two somewhere, reading two, right? So why? Because this is a simple unicast, it will always start from the beginning, but because this is cold, the data is still there. So I can, even if I attach later on, I will get the data, right? So that makes sense. Okay, now what happens if I do a multicast? If I take the same source, right? But just without making it hot, just make it a multicast. But I'm still attaching my second subscriber after 500 milliseconds, correct? Well, even though this system is cold, the stream is cold, once you attach a subscriber, right? In a multicast scenario, it will start giving the data, correct? And if you wait for 500 milliseconds and then attach the next subscriber, that next subscriber is bound to lose some information, right? Even though this is cold, because this is a multicast and you are attaching your graphs, you have two independent graphs and that's why we call them graphs, right? So you have two independent graphs, graph one and graph two, and I trigger them one here on this line and one on this line, right? So because I do this, you will see a slightly different unexpected result. Okay, so you will see that reading is happening for one, only one. So there is no twice reading. So we achieve what we wanted, right? A multicast. But squaring happened for say one, two, three and eventually four, five. But doubling started with six, sorry, doubling started with four, why? Because it got attached after 500 milliseconds and it's completely independent graph, so independent trigger. By then I don't know in future how many subscribers are going to come, so I can't just keep everything in memory, right? Whatever it is doing. So it will just drop after a limit, it will just drop the elements and the late subscriber will not get it. If my intention is to get that all subscribers, right? Should, even they attach later, should see that. Well, what you have to do is create a graph which includes both the flows, right? And different systems, so if you use RxJava, there is a different way of doing that. If you use Accustoms, there's a different way of doing that. So that is what the multicast fork is about. So there is a graph DSL, where you don't trigger twice. You don't say that, okay, this is a subscriber triggered, this is a subscriber triggered. You create a whole graph, like we have done here, and then in the end we trigger. So there's only one trigger, graph.ran. You first create a graph, you create a junction called broadcast, and then you put the sources in the broadcast and broadcast out will go one on this flow, one on that flow. The whole thing, the whole thing is a single graph, and then I have a single trigger, because now I'm controlling my trigger, there won't be any data loss because now the whole graph will be triggered at once. It's not just one linear flow that will be triggered, right? That's a nice thing to have. So let's run this and just verify that works. Okay, so I think this is as expected. So I'm doubling is started with one, squaring also started with one, squaring went all the way till 10, and doubling went all the way till five, right? So this is what we wanted to achieve in the push base systems, right? So if you want to summarize this, multicast is for push base graphs. There are multiple strategies when you have multiple subscribers. You can, if you can, you can give a back pressure. In our case, we could give a back pressure, right? Because source we are controlling, it was a cold source. So we can say that, okay, I have a graph and these multiple subscribers are attaching. And after I click, if one of them slows down, if the squaring flow is very slow, I think it's already evident, right? Because my throttling of squaring, doubling, and the source are all different. So they are working at their natural rate of production is different. But because I am using the cold source, the back pressure is coming into action. And my source is being prevented from emitting more data. And the bottleneck will be the slowest link. In my graph, there are only two links. And the slowest is squaring. That will determine the flow, the throughput. But you are memory safe, because the back pressure will not work only on the linear flows, but on the entire graph flow. Does it make sense, right? But you may not always want to do that, right? I mean, in some systems, you can say that, you know what? My throughput is more important than the loss of data, because it's a monitoring system. So you will say that, okay, drop. And then, so you will say that, I will buffer up to 100 elements so that slow subscriber can catch up if it is a transient difference. If it cannot, it will get fewer items, because I will drop my buffer internally. So there is a buffering, which is happening internally, and it will start. Now while dropping, what can you drop is also, or you can also fail. You can say that, okay, if you can't do that, then you will fail the whole graph, and you will get an exception and the stream will terminate. So multiple approaches are possible. And another side of the story is that, lazy pipelines can be push based. Push based pipelines push on subscribers, right? Like pulled by traitors, this is pushed on subscribers. So they're dual. They're in a way dual of the pull based. The examples are there. They could be hot and cold, right? Both hot and cold, if they're multicast, they can, you can reuse within a single consumption, right? But cold, you can reuse across multiple consumption without a multicast, but then the flow will always start from the beginning. One good property of cold source is that you can have back pressure, right? And then you can have, of course, sink back pressure, that doesn't make sense. If you have a sink back pressure, as we saw in the request paradigm, then that is what reactive stream protocol is about. So that is the, I think then the both sides will look like this. So if you want to just have a summary of the whole thing, it looks like this. So I think that is the major part of the talk. Now I'm going to do three demos of slightly bigger demos, but so far, I mean, is it clear? Is it like? Okay, cool. So let me do a couple of demos now, right? Let's start with the first one. So I have a source of, say, a streaming source of images, right? Files, 10K, I have taken 10K, but it could be infinite. And then I want to see these images. So these images are produced by frames, by framing a video. So if I render them fast enough in the browser, I should get the video back without any encoding of the video, right? That is the objective, for example. So what I do, I put a HTTP server in between because I need a way to, from a browser to talk to, right? And I'm using WebSocket because plain HTTP doesn't work in this case. But on the other side, I'm using a file IO. So this is my source on the disk. I'm on the other side of HTTP server, I'm just reading the file. I'm using the libraries, which I'm using Akka streams, Akka HTTP and Akka streams on both the sides, not both the sides, this side, this is Akka HTTP, this side, I'm using a library to read the data. There is a full back pressure here, which is async. Now, over the WebSocket, once I get the data in the browser, I'm using Scala.js. Have you heard of Scala.js? Yes. Right, it's pretty cool project, and this is the first time I used it. So I'm using Scala.js to render this, but I need a streaming library there also, or else I'll have to deal with events. So I'm using a RX implementation in Scala called Monifu. Have you heard of Monifu, anyone? It's like a RX implementation, just like RX Java, but unlike RX Scala, which wraps RX Java, Monifu can work both in the browser and on the server side, because it doesn't wrap Java, and anything which is pure Scala can be compiled to Scala.js, right? So that's why I'm using it here. So maybe I'll show you the code later on, but let's see how it behaves. Okay, so these are the images, and on the right-hand side, you see the throughput per second. So these are coming from my disk in a HTTP server. An HTTP server is serving it to the browser, and the slowest bottleneck in our experiments is the browser actually, right? So the throughput is, otherwise this could be very high throughput. I will show you in the next demo. But every second you are getting around more than 40 always, if not 50, right? 50 frames. The size of the frame is less than 100k, which is like 50 to 100k, and they are going all the way from the disk to the HTTP server over the web socket and in the browser, and the browser is rendering that. So there is a canvas rendering also happening. That's the synchronous back pressure, which is being applied, right? Because unless I render the first image, I will not, my web socket queue is blocked, right? So I will not get the next message, and if I remove that back pressure, then things will be much high throughput that we'll see. And this is not the encoding of a video. This is just frames which are being rendered. If you just see these are just frames being rendered, and if you want to see now the code, I'll just restart this, so that it stops. And you want to see the code, it's here. So this is my web socket program. I think that is what I like to show first. So it's doing a lot more, but right now I'm just focusing on the images, and the image part is image rendering. This is the logic for image rendering. And this is where the Monifu is helping me. So I'm doing map, map, flat map, map, buffer, and all that. This last line is giving me the throughput histogram, right? So I'm saying what are the events? How many images are being rendered per second? And that is as easy. If you use any streaming data library, it is as simple as that. You buffer all the events in a second, and then you map to compute the size of those elements, and then you print lines, right? And that is what I'm seeing on the browser. So it's as simple as that. I get over the web socket, this is my socket. I convert that socket is a JavaScript API, so it's event-based, right? Any event-based API can be converted into stream-based API using any library. I mean, there's a pattern to do that. So I just do that. If you want to, this is all open source, so I mean, it's on the GitHub, so you can see that. So this is the bridge, which is converting the events into the stream. And then on that stream, I say that, okay, make a URL to render a blob in a canvas. You have to create a virtual URL. So you create that, then you do the rendering, then you say load it, and then so on. So this is the client side. On the server side, maybe we'll go later. But this was the first demo. Okay, let me go to the next one. Next one. So this is similar to what we saw, but instead of browser, I want, on the other side, also I want HTTP server. So my use case is like, I want to read, streaming read of these frames from here, using HTTP on file IO. And then I have an agent in between, right? Which will make a get call. So this is like a browser. But now my JVM best agent, right? So this agent will make a get call. But instead of web socket, now I can afford to do HTTP chunk, because there's a wonderful support in the backend libraries for chunk HTTP. So I get, on a single get connection, I get a stream of images, but like HTTP chunk responses. And that agent will make a post call to a second HTTP server. So there are two HTTP servers. Second HTTP server, again a chunked post. So I will single post, but it will be HTTP chunk. So it will pass the frames. And then this will put in another directory. It will copy it. And I will use the same data. In fact, I'm doing slightly more than the previous one. But in the previous one, there was server, which is reading from the file and coming to browser. Here now browser is agent, kind of. And it is going and again, posting to the another server, which is saving in the file. So almost double the effort, but see the throughput now. So let me just kill this and start again. So let me select one to one transfer frame bytes. So the same 10,000 frames, right? How much time does it take? Okay, all, including the booting of the SBT, the JVM, basically there are two projects. So you can discount some one second or two second. It took almost five seconds and it rendered 10,000. It transferred 10,000 frames from disk to HTTP server to the agent to another HTTP server to the disk back, right? So which is really nice. I mean, this is not the best of the performances. I mean, there are other libraries which will do better than this, but it works very well. Especially on the JVM world, the performance is really nice. I'm not rendering anything here, right? Just transfer. Now I'm not using any serialization. I'm just using bytes. If I use serialization, I will pay a cost. And I have to use serialization in my project. So I'll just show you how much cost. So frame objects are actually serialized objects. I have my domain model which I serialize using a pickling library. And then this will be slow. This will be like double the time at least. So instead of say nine seconds, it will be around 20 seconds. But this is still not bad. This is much better than the, this throughput is around 50. The earlier throughput is more than 100. And the browser throughput is like this is, I think for 500, the earlier one was 1,000 and the browser is 50. So this is a huge, huge difference. There is no WebSocket client library for Akka HTTP. Or else I could have just kept the protocol and tested that, okay, whether it is protocol or whether it is the browser. So this is 16 seconds. And this we are doing for a stream of files. But because now my requirement may not be just for frame of file, it could be just one large file or few large files. But I still want to do streaming. I can very well do that. Because now instead of one connection, I will use as many connection as the large files and each connection will stream the chunks of the file. So that is another variation of the same thing. So if I say one to one transfer block pipeline, then it will, I have four, I have many movies but I'm taking four of them. I'm reading and I'm writing, right? So if you, if I, so there are four reads and four writes happening, but the same flow. It's coming from the disk, but instead of frames, it's happening. And these files are huge. Each file is like around half GB, right? So this is the one. So I think you appreciate this flow, right? That okay, you can write an agent which will do all the end to end streaming. And there is a async back pressure everywhere. There is no synchronous. There is no synchronous rendering like browser, right? So there is a async back pressure end to end. We can do a final demo now where we keep the same idea, but now extend it just to show multicast, right? We extend it. We say that, okay, now my agent will be one but I will read once. I will get, my get stream will be one but my post stream will be two. So that doesn't mean that I will have to read twice. I'll multicast it with the back pressure which means that if this is very fast and this is very slow, the whole throughput will be determined by the slowest link, but I'm fine. As long as I deterministically transfer all the large files because they are the, it's not, it's streaming but the requirement is not about, okay, drop some of the frames. It's a monitoring system. No, I want deterministic transfer of all the data. And if it is slower, that's fine. As long as it's safe for memory and it's streaming, I'm good. So we are going to run this and this is even more complicated. So there are three HTTP servers, the three side file IO and there is one transfer agent with three connections, right? Now, if I run this, one too many transfer. So you will see that you're reading from four movie files, right? And then you are writing eight. Those eight are basically four for the one HTTP post and the four for the another HTTP post, right? So writing, but reading is happening only once. So a chunk of, from each of the connection, the first array of bytes will come, right? And then that array of byte will be transferred to both the post array of bytes and the second will come and then that will be transferred. There is no holding in the memory, right? Because then we only buffer that will be holding, but not much. And the lowest link, the slowest link will give the back pressure. So this was the final demo. If you are interested in the code, I can show you some code and we can talk, but I think we have 10 minutes left. So just let's do a Q and A. All these codes are available. Yes, so if you go to, say, Github, Mushtaq and Data Transfer, you should see that. So if you go to Github, Mushtaq Data Transfer, this is the link, right? And if you go to my page, there is also a reactive streams project, which is the other demo that I was doing earlier, right? So, yeah. Any questions? No, no. This is separate. So which is, say, Github, reactive streams. This is the other small ones which are snippets I was learning, right? Any questions? We have some time left. You want to see the code? We can see the code or else. I did not throttle them. So they were possibly working at the same rate. But if one was slower, that was my throughput. It will be kind of that. I mean, so if you are using JDBC, then you are not getting any non-blocking thing. You can convert that. Even JDBC with JDBC, you can. So if you don't require the true non-blocking thing, you can wrap the JDBC into a streaming API. And you will have to deal with a separate thread pool. So a thread pool will be separate for those IOs. But it will still be streaming over HTTP, sorry, or JDBC. You can do that, but it will be blocking. But it will expose the API of a stream. So I think if you slick is one of the libraries, right? Which tries to do that. So slick gives the, it's like a, like hibernate, like hibernate, but very different. If you connect to even the blocking JDBC drivers, it gives you the custom handle. But it gives a separate thread pool for that. So it's blocking internally, but it projects as if it's non-blocking. So you'll have to do something like that. But you are right. Yeah, yeah, they do. What do we have for Cassandra, do we have something like that? I don't think Cassandra has a driver. Cassandra has main emphasis on Spark. So Cassandra, if you want to, because Cassandra is in a distributed world, if you want to process data on the same topology, then you have the similar Spark cluster on that. And then that works very well. So the more focus is on that, rather than say connector for getting in the application, because that is not the best use of the topology of anything else or should we, I mean, done? Yeah. Cool, thank you. Thank you. Thank you.