 All right, everyone it looks like we've got some people here already. That's great This is the first time I'm trying to do this sort of multi-stream business in theory It seems to be working like I think people on all three channels can both see me Hear me and I can see your chats and you can see each other's chats If things start going wonky then just let me know and I'll try to find some other way of doing things In case this is the first time you're joining me live. I'm John I do a bunch of this sort of open-source streams and particularly in Rust I have a patreon page and also a Twitter page that might be where you found me If you want to hear about upcoming streams or discussions about streams I'm considering doing or even just to find videos I've done in the past Then feel free to follow me on patreon or on Twitter and you'll get all those announcements I also have one thing to point out from a last time stream So part one of the stream we did Tokyo Zookeeper and we did sort of the async Primitives and the future primitives that were needed and after the stream someone opened this github issue that I only saw earlier today The pointed out that if you remember back in our stream, we wrote this sort of wrapper for turning the The raw socket that we have the async read and async write into a sink in a stream So that we could send things back and forth and it turns out there's actually an official sort of Tokyo crate for this And that's called Tokyo codec And if you read Tokyo codec It basically is an adapter to go from a stream of bytes into frame streams implementing sink and stream Which is basically exactly what we ended up implementing ourselves now that said I think it was actually a good idea for us to implement the primitives ourselves just to see how they work because I think that's more something to learn from but Just so you know this going forward if you have to implement this yourselves and also if you want to like try something You could try to implement Sort of change over the implementation We wrote turn it into something that uses Tokyo codec and I'd be happy to merge it I might get some time to write this myself But at least now you know that there exists a less slow level way of doing these things If you have a question feel free to just like ask it and I will try to monitor the chat and and answer whenever I can So Zookeeper so Hopefully most of you have seen part one of the stream even if you haven't though I think it should be possible to follow along so in part one of the stream We built sort of the the core code that's needed to communicate with zookeeper So sort of the serialization and deserialization. How do you do that in an asynchronous way? Tokyo IOPL. Oh, yeah Tokyo IOPL was pretty cool So this is a smaller side, but I wrote a Might as well discuss this here. Why not I wrote this thing called the Tokyo IO pool, which is basically so Tokyo is great, right? It lets you do a synchronous thing and have a pool of threads that operate on them But it has this one limitation currently that is all of the IO goes through a single thread So all the futures execute on different threads and this works dealing. It's very very fancy and efficient But all the IO goes through one thread, which means that if you have a program that is syscall bound Or or it's this very IO heavy You're gonna end up seeing that one thread become a bottleneck And so Tokyo IO pool is a slightly different set of for Tokyo where you run many threads that each run their own Single threaded Tokyo runtime with their own IO reactors So the reactor is what Tokyo calls the thing that does syscalls for you essentially And it just runs a pool of things each with their own and and then when you spawn a future onto that pool The future is sent to one of the threads and then executes on that thread forever more And this means that you now get to have all the threads do IO in parallel And so if you have an IO heavy application, you might want to look into this It still does have some shortcomings such as there's no work stealing There's also work in I'm chatting to Carl the maintainer of Tokyo to see if we can merge it into upstream Tokyo Which would be really cool But for now, this is something that may be interesting to look out for What do I think of the now stable SIMD modules I haven't had a chance to look much at the the now stable SIMD stuff I know that SIMD is really cool You can do some really efficient things for it But it's mostly if you're doing like data analysis or math machine learning maybe for the kind of stuff I do It's likely less relevant So the my current research project is working on building a new tidbit database And it's a little unclear how we would optimize that with SIMD because we don't do like matrix multiplications or the kind of operations Where you very clearly benefit from SIMD Rust man, is that me? I'm gonna assume that's me How do you think of Haskell and Monads and also the relationship with rust? I mean, I think Haskell is an interesting language. I think it can my limited experience with Haskell has been that it can be Tricky to build things that operate with the real world like if you're building something that like needs to talk to like network drivers or you're implementing like I Don't know things that have to interact with the user has closed really good for things like parsing For compilers to some extent although camel has an advantage there, too It's like Haskell has its niches where it's really cool and it has a really good type system arguably one that's better than rust in many ways But I also like the low-level memory control that rust gives you and Haskell does not give you that at all If you need to do performance debugging Haskell for example You have to go through a lot of pain Monads, I don't know exactly how to compare with rust because Monads are or a programming concept. They're not so much a language There's been some discussion about whether you can implement Monads and rust and I think the consensus is not quite yet But you can get quite far All right, let's dive into what we're doing today So we're looking at zookeeper and what we built as I mentioned last time We built sort of the core infrastructure for talking to zookeeper We didn't actually implement any of the higher-level zookeeper primitives. If you remember Let's pull up the code here So the only thing we implemented on zookeeper was the exists method and nothing else and we tested that it in fact works The exists method will talk to the server ask for a given key And then we'll be told back whether or not that key exists And we found that that indeed works correctly and it returns you the some statistics about that entry Of course, there are a lot more things that you might want to do with zookeeper So one of the things we might have looked want to look at here is the The zookeeper crate that already exists that is synchronous But if you look at zookeeper here, you'll notice that there are a bunch of other methods that we probably want to implement I think we'll probably start with create Create adds a bunch of things like ACLs creation modes and those kind of things And then once we have it created create then we can test exist easily Then we'll probably want delete and once we have that sort of trifecta then we can now write fully self-contained tests Which would be really nice Then once we have those I think we want to move on to The watchers so watchers. I recommend that if you haven't already you take a look at the zookeeper's programming guide Oh Yeah, that's true. So when we get higher kind of type then we have like full monads But it'll take a little while before we get that that there are a lot of complications with higher kind of types I Recommend that you take a look at the zookeeper's programming guide. They have it's pretty long, but also quite readable The thing I'm talking about mostly is these zookeeper watches. So the idea of a watcher is that you can tell zookeeper Give me the value for this thing and notify me when it changes so imagine that you Try to read who is the current leader in a cluster for example and You will be notified you will be told who the current leader is and you will receive a notification Whenever that changes so imagine that server goes down for example This interacts well with things like ephemeral nodes in zookeeper, which we'll also get into But so so having some general idea of the zookeeper data model is going to be useful We're gonna start though with just Create and delete. I think that's a reasonable place to start As always if you have questions post them in the chat and I'll try to monitor that on the side with most of these like I Am also sort of just doing this on the fly so if you feel like you see something out of a way where you're like, hey, that's probably wrong or you feel like You're wondering why I'm doing something a particular thing you should ask and then it might be that it turns out That's a stupid way of doing it and I'll do it differently So let's do a quick sort of recap of the internals that we have so if you remember The way we set things up was that we have this packetizer thing and the packetizer is a Future that we spawn onto the pool that just runs forever And what the packetizer does is it manages all the packets that we send and receive from zookeeper so we open a single connection to zookeeper and we spawn this packetizer that's always running that will that has a an incoming channel for requests and then responds on that channel with a Channel of its own that it will respond with the answer on so it's a slightly convoluted explanation, but let's look at it here So the packetizer that we have Yeah, so when you create a new packetizer What it does is it spawns the packetizer and then it returns this in cure thing and an in cure is Basically just a channel What is that channel? Well that channel is One of these it's a sender of a request. This is a zookeeper request and a Channel on which to send the response So remember that the zookeeper protocol is sort of asynchronous You can send a request and then you might receive a response much much later And there may be other responses in between and so this packetizer operates as sort of a multiplexer Where it gets lots of requests in it serializes them sends them over the wire and when the responses come back It sends them forwards them back along some response channel And so the way we can see this being used as an exists for example It does dot and Q and dot and Q Down here creates a new one-shot channel Sends the request and the sender on that one-shot channel to the to the multiplexer and then Gives you back the Receive end which is where the response will eventually be sent so for for exists What we do is we and Q an exists request and then what we get back is a future that will resolve to the response and Whenever we get the response we just parse out whatever we want Distro and WM I've answered this I think in one of the YouTube comments for another video, but very very briefly. I'm running arch This is X monad although. It's just a tiling window manager. So it's not terribly important The bar at the bottom is poly bar and all my config files So you can go there and look at them and If you have questions or whatever you can also post issues here if you feel like there's something you're wondering Why is there it's probably just like old garbage that I haven't cleaned up Yeah, so so that's sort of the infrastructure that if you ever want to make any API call you Sort of send a request to the multiplexer the multiplexer does some serialization Does some other stuff eventually gets a response DC realizes it and sends it back along a channel Which resolves the future that we got back from in Q And so if we now want to implement say delete Then We'll give some future. It's a little unclear. I should would delete should return here. It returns nothing. So let's make it return nothing And error is going to be failure Actually here Let's do it returns a bull So it returns true if the things was if the thing was indeed deleted and false if it did not exist and It's gonna look very similar the reason I do delete first is because deletion is very straightforward Version version this but I assume we're gonna need it Yeah, so the reason I do deletion first is because if you look at create it takes all these other arguments So we're then gonna know how to serialize and deserialize And so I just want to do the easier one first just to refamiliarize ourselves with With how have sort of the setup of this so we're gonna dot in queue a proto request delete and The question then is what goes inside of a delete well So if you remember from last time what we did last time as well was we sort of Salvage if you will Code from the serialization of rust zookeeper. So if the synchronous implementation Proto, I think it was So here if I remember correctly a yeah So in the old the old zookeeper crate has a separate struct for every type of request I've opted to make it an enum. It might be something that I come to regret But I think for now it's gonna be fine And if we decide to change it then we decide to change it Oh Proto request So we're gonna add a new type of request and that is going to be a delete request They do not need to be pub Aha, so that's where the version comes in great And then of course we're gonna have to implement serialize into so this is if you give a have one of these requests How do you turn it into bytes? And so for most of these the implementation is pretty straightforward It's gonna take a path and a version and it's gonna write out so here We can also sort of take inspiration from the original Implementation here, which says that you write out the path and then you write out the version as an i32 big Indian Yeah, so we first have to write the opcode So this is sort of just a general protocol for zookeeper that any request we send We first have to send the the code for what request we're making So in this case that will be a delete we'll then write out the path and then we'll write the i32 as a big Indian 32 big Indian Version So path dot write to works because we've implemented This write to trait which is basically how do you serialize one of these? and we implemented it for For strings and basically the encoding is as this is very common in binary protocols You print out the length of the string and then you just write out the bytes of the string And so that's the same thing we're gonna do And that's how to serialize a delete Of course one difference here She that's a good question So if you remember from here delete in the original zookeeper takes an option i32 It's a good question. Why does it take an option if the protocol dictates an i32? Unwrap or minus one ah, so the version is oh interesting Yeah, so this is where we're gonna want Structured return types so Notice that if you call delete there are multiple possible error situations It's the thing you tried to delete does not exist or the thing that you tried to delete has a Does not have the version that you passed in Or if you look here if the node has children so in zookeeper all nodes can have data and can have children And you're not allowed to delete node that has children And so here if it has children it will return this not empty business And so this suggests to me that we Actually need a structured return type for delete, which is a little awkward But I guess we can do it so we error.rs Use failure So what we're gonna do here is we're gonna return not just a failure error, but a Mod error and then I guess Pub use error Or no, just pub mod error And this is going to return not a failure error but a error delete and deletion error So here if you remember we're using the failure crate for Writing out errors and the failure crate houses really convenient. Where's my home page? So the idea is that you can derive fail for an enum and That more than it implements various traits that are required to use it as a as an error And then we can do this So we use fail to implement display for each of them in our case we have no node Which contains no information if I remember correctly so this is Which we call it target and target node does not exist We could also return bad version and my I assume that it will tell us what version it is It might not but bad version I 32 Splay Target node Has different version has then expected So here I guess we can be even more helpful and say expected is I 32 and Got as I 32 and then then the really neat thing With failure is that you can also use syntax here to write out Fields from the struct in its display the failure credit is just really convenient for creating these errors that are then nice to Work with no node display text target node does not exist. Thank you. Thank you Yeah, so in our case what we want to do is we want to say a different version then expected and Then we want to use Got and expected Like so and it can also return. What was the other one we saw like has children Yeah, so that is Not empty. It's nice here to use the same The same error types zookeeper has so I'm trying to see if we Apache zookeeper error So this is we're now looking for the original zookeeper source and if I remember correctly, there's a List of all the error types Main Apache zookeeper Util maybe no client no Zoo deaths, okay, so that has all the op codes permissions IDs Zookeeper job I assume not Where does it import errors from? Off result maybe Or proto off no off results. I specifically want to find the Definitions of all the errors because I want to use the same name for the errors as is used in the Java types because when you have these kind of Cross language implementations where all the libraries are implemented differently It's really nice if there's at least some similarity between them So that if you search for an error that occurs in one language, you can sort of apply it to a different language as well And that's sort of what I want to see here Zookeeper It's unhelpful search for No node Seems like a promising Search the question is There is it Find really I guess bad version maybe as a Keeper exception. I could be it Yeah, so I have that Where did we put that proto error? Yeah, I'm just trying to find where I got this from Don't actually remember I Think I see the I have this but I think I copied this from the Rust zookeeper implementation and I just want to confirm that that is indeed what they're called in Here No node exception. Yeah That that is in fact what they're called upstream as well. I Assumed that the same but yeah, exactly So this is where all the zookeeper things are raised and so no node that seems right and this is just called bad version So the question is where is bad version bad version exception? Great, okay, so that it's just called straightforwardly and it's also not see code bad version That's interesting. So there's a code somewhere here maybe mapper Yeah, so it seems like these are the kind of responses that can be given Exactly. So keeper exceptions happen internally in zookeeper Whereas these are the status codes that the zookeeper server can give back So here we see no node, which gives not found bad version. Yeah, okay So it is indeed bad version and then I assume this is called not empty. Yeah, okay, perfect So for not empty Here we say target node has children and cannot be Now some of you may wonder like why don't I just make this have the error be a What was it proto error? ZK error So ZK area is this whole list of all the possible zookeeper's exceptions. The reason I don't want to do that is because This list has more errors than you should expect if you do a delete, right? So if you do a delete you shouldn't expect to get like invalid ACL and you shouldn't have to check for it We sort of want to be able to take advantage of Of match And so there are a couple of ways we could do this We could say we could either say that the item is a result This Sorry this delete and Then the error is that there's a protocol error so if You can think of this as either there's a protocol error or you get an okay or a zookeeper result So it's sort of like a nested error case I think what we're going to do instead is to let this be error Error delete and then the caller would have to match on it in our case Look back at error Not empty does not take any parameters, but there's also one more here, which is like protocol error Right and here failure has this really neat thing called cause Where is a cause documentation somewhere deriving fail Deriving display strings tuples in a Where is here we go cause Yeah, this thing So you can also do this and then say the deeper down There's a failure error and then you can mark it with a cause Now I don't really want this in theory. I don't want this to be failure error. I want this to be Whatever the NQ gives back so let's look at our proto mod Where is the NQ that gives a failure error, why does it give a failure error What do we get back on the send we get a zk error? right so Hmm Yeah, so we could make this return a zk error So when you accuse and cue something. Oh, no, no the item. Oh, I see so here we did the double layered error case of What you get back when the NQ succeeds is either an ok or zk error. Okay, so that does actually capture this So maybe we should just continue using that then so this would return a result this Error is failure Right, so the explanation here is Because NQ already gives us this kind of wrapping we might as well continue it And so here the way to think about this is we're gonna take a And then actually we'll take a Yeah, we'll take an end then so this is going to be What we get back from NQ, so if I remember here a NQ gives you back a result respond a result if it results correctly or an error if there was a protocol error So in our case the protocol error will already be propagated Then we're saying when we get this result back what we really want to do is match on R Because R can either be proto response exists Sorry delete which at whatever we don't actually know what's in a delete response yet. We'll have to deal with that or it could be an Error zk error no node and if we get one of these what we want to return is an ok and then an error delete no node Right, so notice here Sorry Well, it's an ok error This is going to end up returning an ok ok and then whatever value we end up putting in here Not empty is going to be transferred to a not empty and bad version Is going to end up being converted into a bad version? Right, so what we're really doing here is we're Converting that things that we know we could get back from delete and turning them into something the user can match on and then We see if we get any other error, then we want to just return it as a As a failure of the protocol because it's not a response that we expected. What about using bit flags for errors? What were you thinking about using how would you use bit flags for errors in this case? so remember that the the error codes are dictated by zookeeper already and We want to the we want an enum of the error because it might also contain data in this particular case. It doesn't Because it looks like bad version does not come with data, although we could imagine that it did in fact, so I think we're out to It's too bad Keep this around just in case is a way for us to get the versions eventually I Guess at the very least we can include expected It's not how expected is spelled So we expected an i32 and we at least can print what we expected But we may not know what the Code of it actually is Yeah, so what we're saying here is that we get an okay everything is fine if we If we get a no-node then we propagate that as a there wasn't a protocol error But there was an error to your request Similarly for not empty and bad version and any other kind of error that we got from zookeeper. We turn into a protocol error Missing fields Path and version right so we need to give it path is gonna be path I guess And I think in this case Here we're taking a string we could be nicer here and say we take any type that can be turned into a string I think this is fine for now Yes, we can call this to string Hello and then for version remember from the zookeeper crate it says Here they're unwrapping it at turning it to minus one so my guess is that Minus one is some kind of special value It looks like the response is actually empty to a delete so that's gonna be useful to know too So here in fact no node You won't get known well So here we also need to decide whether we want if you try to delete something that's not there is it an error That's a very good question So this will be an empty response Because there is no delete doesn't have any content. And so the basically what we're deciding is whether we want this to be This So this is a totally valid way to expose delete But this is also a valid way to expose delete Right do we want Do we want trying to delete something that's not there to be an error or to just return false Think we might want it to be an error just for Consistency Yeah, let's do that Okay, so in that case the result Do you get back as an item resolves into nothing? So if we get a if we don't get any kind of error then we succeeded we get no node not empty or bad version We did not succeed Yeah, I think that makes sense Now let's see So version is going to be version unwrap or minus one my guess is that minus one is a special value that indicates You don't care what version it is In fact, this is one of the things that bothers me a little about zookeeper is that the protocol is extremely poorly documented Let's see if we can find The docs for the Java implementation API docs. Oh I hate these kind of docs zookeeper client No zookeeper Zookeeper Methods delete Ha Yeah, so they've opted to make no node be an error which is basically the same thing if the given version is minus one It matches any nodes versions Okay, actually we should probably at some point what we should do is take all the documentation. That's from the That's from here and stick it into our library to make sure that we Sort of expose the fact that Or expose any of these kind of things right like the the Java API is documented to say what happens when you do various things But let's not deal with that for now Why is it complaining? That's the real question All right, so we're gonna the things we're gonna have to do is we already said how to serialize delete we're gonna know how to Going to have to know how to deserialize an empty response So the response to delete is just you didn't get an error, right? so in our case and my guess is this will there will be other Zookeeper commands that also have the similar kind of mechanic where you just get an empty response back if there wasn't error So here empty And We have any of these right so if you get a delete and you get an okay Response So this is saying if you sent a Yeah, if you send to delete then the response we parse out is we just don't parse anything more One question here is whether we'll still get errors. I think so So let's remind ourselves what we actually do in the packet iser When we pull read We read some stuff in oh that This e-print should probably go away We read some stuff from the underlying thing We read the length Here we look at the first thing Right, so we look at oh, that's right. We read out the header from the So if you're if you recall from the last stream the response we get from zookeeper includes is like header that contains the ID of the request that's being responded to the zookeeper version ID, which will come back to a lot later the error code if any and Then followed by the body of the response and so really all we're doing here is we're checking what the error code is And then there will be no bytes left so here when we decode the response to a delete We know that there are no more bytes to read and so we know that the response is empty in theory We could do something like Assert that the response is empty But let's just not do that for now Also, let's I guess we can keep these aprons in it's not they might come in handy if we probably have some bugs still So what are you building right now a zookeeper client and rust yes, so I'm building a an asynchronous zookeeper client and rust So I recommend that you watch if you're interested in sort of the asynchronous parts of things I recommend that you watch part one so part one was very focused on the low-level primitives and now we're building An asynchronous API on top of that to match the features the zookeeper has But part one was a lot more async oriented So it depends whether you're more interested in zookeeper or more interested in async Although this part will also be a lot about async Why should deleting what doesn't exist be an error at all? So it depends a little bit on your application if you Expected that thing to be there so you could delete it You really want to know that your deletion failed because that's actually a bug in your programming logic somewhere, right? But you're totally right. There could be some cases where you Delete something just to make sure it's not there in which case you don't care about the result Mmm But it's important that we provide some mechanism to figure out Whether or not your deletion succeeded because some applications care and some do not the ones that don't care Should just ignore the error if one occurs, right? And now we give them a sort of a complete mechanism for determining this you do the delete If it results in a protocol error, then like your connection might have gone away in which case you'll have to do something If you get an okay, then you know that you've deleted it And if you get an error you can choose what to do in response to that like for example You might ignore if there's no node, but if it's not empty or bad version then In all of these cases the node was not deleted And that might be information that's useful to the to the programmer Let's see, what is it complaining about type mismatch Right See, that's a little frustrating. I think we are gonna have to use then here. So so and then is Take the successful response and turn it into a future Actually, no and then should be fine turn it into a future whose Okay value is whatever you want and whose error value matches the previous error So this means that if if what came in here was an error e of some type e Then and then has to return something where the error is where the error type of the future is e And what the compiler is complaining about here is the e that we get back here So the the error we get back from in queue is a zk error. So e here is of type Proto error zk error But the e that we return has to be of type failure error And so that's what the compiler is complaining about and it's totally right There are a couple of ways we could do this What do we do here? Right and exists we just Bail which is probably what we could do here to to be honest Although it might as well be format there. I don't know why we made it Because in this case we just want to Propagate back the error. So we want to say delete call failed With the error now again here We could do slightly better by using context and sort of wrapping up a failure And this is probably what we'll end up doing later to make this API nicer in a sense is to take the The errors as they occur on lower levels of the stack and keep wrapping them in more context So that at the time you get up here and there was a protocol error You can actually figure out exactly where that error came from but for now Let's just use format error, which is expressive enough for what we need Bad version right so bad version we also wanted to include expected which is going to be version So here let's do version is version dot unwrap or minus one That way that type is now copy and we can move the type into here Match arms Requests line 117 All right, the opcode of the delete requests is opcode Delete this is why Pattern matching is fantastic because it told me I forgot to implement that And then 81 it's saying The other patterns are not Oh Yeah, so this is the this is one of the things is a little bit annoying with our current API and that is the Because the response is an enum We have to be prepared prepared here to handle any other kind of response type And this is probably gonna end up being really annoying like We know that there's just no way you can get back like a Exist response to a delete. It certainly should never happen. So I mean we could do this Non-empty response to delete Ideally we would want to fix this but it's actually a little bit tricky because we could give like a t to nq but keep in mind that the Where's the here The sender that the uncure has is just send a request, right? This is not generic over t and sort of can't be because you would need one channel for every type of request Which is not what we want to do And so I mean the in cure would have to like take a t Where t is something like into request and Then it would was the result to be something like t I Guess it would also be generic over r where r is the response type But this is an r and now the problem is what you get back on the channel of course is also not generic It's an enum and so it basically would mean that the matching gets moved into nq because this would have to Take your request rapid and enum and then unwrap it in the response Though maybe that is what we want. It's just a little bit It's a little bit sad Hmm Basically the the question we're making is do we want to keep tagging things as unreachable in the callers thank you Or do we want to wrap up the logic of enum wrapping and unwrapping inside of nq? And I'm not actually sure which is better What double dots means where oh if you're this you're talking about somewhere up here, right? I don't think I have a double dot here anywhere, but I think what you were talking about is if you do something like This the double dot that's there this means match if Match this if it is this variant ignore all the fields Similarly you could for example bind only one field and not the others dot dot is like ignore the other ones You can think of it as underscore, but for more than one field Alright, so in theory, I think that should mean that we have delete So this is I don't know if you remember but at the end of the of the last The last stream we talked about how now that we have the infrastructure in place in theory Adding all these additional API's on top should be very straightforward because we have a very nice mechanism for sending requests and getting responses mmm And if you look at our implementation of delete here, there's very little there that is not related to delete And so we've managed to sort of encapsulate the the protocol quite nicely And so of course the way we're gonna test this is that we will have something here like we're gonna connect then we're gonna Exist then And then ignore the response and do No, actually this gives us What is that? right Then we want to zk. hmm delete through with none and then Exist again. Oh, yes start and then underscore Here this is just empty So this will just be deleted and Then here we're gonna do zk.exe So this is basically currently this isn't even really a test all it's doing is it's doing exists then deleting then doing exists again So let's start zookeeper And now let's do Cargo T So this is more for our own sanity to see that the the basics of this at least works. Oh Yeah So this is one of the things that's a little bit annoying with with futures is like threading along zk So we have a couple of options here Because remember zookeeper up here and theory at least is clone We could make more of the zookeeper thing and so we could totally allow You to do something like that zk is zk clone Or zk like one and then we could move Zk one into here and then we could also make a zk two and then we could move And then zk two into here So this would totally work at least in theory And in cures already come Right, so it doesn't complain about this and so that's all totally fine This is a little bit annoying though, of course because you have to clone it for each Thing you're gonna need the way many libraries get around this is that you have all of your futures resolve to The instance that you were using followed by the response Right, so this would be The stat would be the second thing and the zookeeper with the first or vice versa So you can see this a lot in things like Futures If you look at futures Sync mpsc sender Sender so that's sync Is it not sync where this is done? This yeah, so if you have a sync so any kind of channel for example It gives you back a future called a send and if you look at send its item is S which is the sync so if you do some channel dot send foo then When that send resolves it gives you back the channel again so that you can send more stuff And that's basically the thing we could do here Of course, it means that you have to keep matching over. I think we'll still want to do this I don't have a better pattern for it Which is a little sad Here we don't care about either here. We want to get the zk back And now this Cloning can go away and we don't need to move does this make sense? So the maybe it'll be easier to show it up here So what we're basically saying is that the item that you get back is always going to be a self and one of these Now we could of course make it consume self, but there's no good reason Similarly down here the item you get back is going to be a self and that and so now we're gonna do here is dot In theory, you should also then return with the error which gets annoying But I think we're just gonna map We maybe have to clone in here. I don't want to do that actually take it back keep it the way it was We still want to keep her to be It's a little unclear actually whether which API we think is nicer this cloning is really annoying The reason I don't want to do it the other way is otherwise If you call exist we would have to clone self regardless So that we could give it back because we get self by reference and then we have to give it back at the end The alternative of course would be to say that exist consume self and returns Self when the future resolves, but now like if there's an error then that gets sad Although errors are protocol errors, so maybe it's okay Ah, it's a good question. I Think like the most future-like way is to do this because it's what channels do But it is a little sad Right, so basically that's all we're changing it to So and now down here. We don't need the clones ZK is consumed by calling exists And is then passed back when it resolves This ignores all its arguments this gets the ZK This is a ZK and this looks at the stat and now of course ZK nothing What I mean is the leading and non-existent entry is basically no-op success On the other hand checking existence by deletion is kind of awkward code path So This is why we were discussing whether it could be that we should have this return Sorry return a bool and then have no node not be an error, but just return false We do there does have to be a way for you to check whether the delete succeed or not So it has to be one or the other we couldn't just like Silently treat the case of no node because there are applications where if there wasn't a file there before But you expected there to be that's an application error and you want to handle that So we do need to expose it in some way. The question is what is the best way to expose it? I? Suspect that in general if you try to delete something you're expecting it to exist But I am not sure Therefore given that I don't have a strong preference I would like to keep it the same way the job API is just for consistency for people are familiar with both That might not make sense, but it's my inclination I don't think it particularly matters the implementation is pretty trivial to change one way or the other One thing you could look at is so think of file APIs, right? So if you look at file In the standard library, we could try to be consistent with that and if you call file Where's the I guess it's just an FS probably Remove file Yeah, so remove file returns a result where the success is nothing and Return an error if That's unhelpful it doesn't actually say Source Okay, so just calls unlink That's unhelpful It's a little surprising that this is not oh Yeah here, so if you look at remove dear removes an existing empty directory So the implication of remove is that the thing has to exist and then the this return type sort of tells us that If it does not exist you're gonna get an IO error and in particular if you look at it IO error has this thing where is error kind error kind Called not found which is basically the same thing as what we're seeing in zookeeper If you try to delete something that's not there you get a not found exception or error So I think we want to preserve that that's similar semantics Okay, so our test succeeded Although it's not particularly helpful. Let's zk cry and Then this is just connecting to my local zookeeper. I'm trying to what is it? Create Slash foo Create such food data Great, so now we created a node slash foo it contains the data data And now if we run cargo T. Oh Did I do something weird? Oh, right? I was just doing that We now know in cargo T What we should see is that the first Exist call should get that there's a file there then the delete should succeed and then we should get the second exist Call return does not exist. So let's see reading through all this. We got the connection. So that's fine Then we get the request exists We get some response and so exists return some and a stat Which contains the information about the file now exist doesn't actually give you the contents of the file But it does say over here data length four, which is the length of data indeed And then we do the delete request It handles a response We see the deleted message, which is what we print when a delete succeeds, right down here Then we just print out deleted which is indeed what happens here. We print out deleted Then we get the exists and the response to exists is not perfect So this means that both The original exists and the deletion and the second exists work correctly So we now have almost all we need to have a self-contained test The only problem is I currently still have to run This create food data myself to set up the node and so let's go ahead and implement create as well We already have a structured error type here actually in the sense of option Because for exists the only thing you can get back is either to the node exists or it does not anything else is a protocol error so we now want create create is also going to take a self it's going to take a path and we know from looking at the Existing implementation that there's also going to be some other arguments But let's ignore those for now It's going to return self and then something here that we don't quite know what is yet and it can also produce protocol errors Like so and we already know that the setup is going to be very similar to delete and all the other All the other things that we've seen and so create here Start there. Let me check something When we send out a packet Sorry, I'll get back to what that was later All right, so now let's look at what create does so it's probably going to be somewhere here Create create create create this thing and let's see also what the rust version does So the rust version of create here takes a path. It also takes a bunch of other things Right, so because we're creating something we obviously also need to give some data That data That's a good question It's gonna have to be owned because we're gonna have to it's gonna have to be static So we'll have to send it somewhere. So there are a couple of ways we could do this I'm a little bit tempted to make this a cow static So imagine that you wanted to do something like zk.create and you want to create dash foo and then you want to create it with some Static string like hello world. This has type static u8 Right and it's totally fine for you to pass that as the argument However, if we here take a vec q8 So then now if you wanted to call it with a static string, you would have to do this You would have to do something like that from this or something which ends up copying all the bytes It would be nice if both ha like if we take the alternative of having a vec it would be nice if these These top two both just worked out of the box and it turns out there's actually a way you can do this And the way is to take a D where D implements into cow static u8 so a cow is a copy on write data structure that holds either owned data or a Reference to that data that can be turned into an owned one so in this case what we're saying is either give me a something that holds a static reference to a list of u8s or Give me whatever that can be turned into an owned version as and in this case. There's the Rust So cow operates on the to owned trait So if you look at to own you see that there's an implementation of to owned for slices And that is gives you a back and so in our case Vec implements this because you can just make an owned cow directly here you would just have an owned Vec u8 and Static strings sorry static byte strings would also implement into cow because Where's it here? Any T that can be implemented as as ref implements into cow so in this case This would just be turned into a cow borrowed with a static lifetime a slice to a u8. So both of these are fine So This data is gonna be anything that takes a D. What is it complaining about? Why am I I have a syntax error somewhere? Where's my syntax error? future Where and of course these have to go away perfect And we're gonna have to use cow use I think it's in borrow ACL we're gonna need and mode we're gonna need but for now. Let's just say that we're gonna make a Create request. It's gonna take a path We could do a similar thing with strings and say that we want anything that implements cow string Actually come to think of it that might be what we want to do the it depends how much we want to cater to Static things because you could totally imagine that Static things are not usually that long and so therefore it probably doesn't matter copying them The reason I did it for data and not for path is because a path is just like not going to be long But you could in theory imagine that someone like does include string or something to include all the Contents of a file and tries to write that into a node and that we don't really want to copy this is definitely a Mostly a performance optimization although it does also make the API a little bit nicer because as you saw it means you Don't have to explicitly do this back from business So we may want to do it for paths somewhere to We're gonna have to use cow. That's true. It's pretty funny Okay, so our creation is gonna take a path. It's gonna take a data word. That's gonna be a Just gonna take data It's also gonna take this ACL and mode and if we go back and look at them Where's this? So we're gonna need ACL and mode So an ACL just has a bunch of permissions Create mode is just how do we want this node to work? So here we need to know a little bit about the zookeeper node types So a persistent node that's created if the client goes away that node still is there So it's not deleted when the client goes away as opposed to an ephemeral one an ephemeral node when you create it goes away Persistent sequential and ephemeral sequential are sort of like the two above a sequential node Has a sequential number appended to it automatically by zookeeper. So the name if I you create like Slash foo and you create it with either of these sequential nodes It would be created a slash foo one and then if someone else tried to create a slash foo It would be created a slash foo two and then etc. And the reason you want this is to implement things like hues Container I don't actually know what is oh Interesting. Oh, well, I mean I guess we might as well implement all of these There's a question of whether we want These to be different enum types or you could imagine persistent having a Boolean inside it or a field called sequential Or you could imagine that you had sequential be a Container doesn't apply. I think we're just gonna we're just gonna stick with the same thing that is done for create mode And so here Create So create mode Here there's a proto. I sort of want I guess it could be a request Um, let's see. So create mode Can be one of these guys and they clearly have like predefined values that we're gonna end up using in the protocol um So that's all fine And I guess we could say that this probably has My guess is it has a wrapper of i32 But let's double check So in zookeeper remember how there's this uh jute file That defines All here Defines the entire network protocol. So let's see if we do a create request Where's create request? The eight request So there's a path It's a buffer of data acls and then flags And flags is an int Uh, which is an i32. So what we're going to do is we're going to do the same thing as we did up here and say uh here The wrapper i32 this means that all of these will already be an i32 and we can cast from create modes to i32s Just by casting um Oh, I see. So if you if you look carefully here, what is really happening is that uh, this is really just a flag So if you think of this in terms of binary flags, um Could actually use bit flags for this in theory although this sort of This implies that you can't have a an ephemeral container Which it's actually unclear whether you can uh, but if you think about this as a bit string of three bits So the uh first bit So this bit is uh Is it a container? Right. So remember that this these correspond to four two and one This bit is uh, is it sequential? And this bit so the low bit is is it ephemeral? ephemeral Right, so if you if you look at the the integer values that have been given here Zero is all of them are zero. So it's not a container. It's not sequential. It's not ephemeral, which makes it persistent Uh, this has one so it's ephemeral, but not sequential or container Uh, let's go to this one. For example, this one has both one and two set Which means that it is ephemeral it is sequential, but it's not a container and this is four So it's just a container or not the other things. Um Uh, it is pretty tempting to make this a bit flags actually Uh, so for those of you not familiar with bit flags Uh, bit flags is a neat crate that basically lets you define different flags and then people have a bunch of operations they can do on them Um It's a little bit tempting. I think it's mostly an api optimization There aren't that many flags at the moment and it's not even clear that it's a full flags because I don't know if you can Create a container that is ephemeral for example or a container that is sequential I suspect you probably can but it just doesn't make that much sense Um But that's why I think we should just stick with this being an enum for now And then we could optimize that later if we wanted to um So Here create mode So here we're going to do pub use Oh, there's a types. That's probably where I want this to go Uh, 30 db Uh, source Yeah Perfect Uh, so we're going to use types stat and create mode Okay, so that deals with create mode It's going to be a mode. I guess we can we might as well go to requests here and add the uh I want to move this up a little So we now also have a create request to something that can be made The path is a string again here. We could use cows To prevent some copying but paths are probably short anyway Uh, and we want to use borrow cow And it's going to take a data which is going to be a cow static u8 So any static anything that can be turned into a sort of turn into a static reference to a list of u8s um We also have what else was there mode Which is going to I guess here we're going to use create mode So we're going to have a this is going to be a create mode And we're also going to have a acl in here, which is going to be this beck of acls So we haven't quite figured out what that type is yet. My guess is we're going to have to use it All right, so we got to figure out what this acl business is. My guess is just a list of fairly straightforward permissions Let me look at it here. Yeah, so we have we give a Right, so create takes a list of acls And the acls just have a permission whatever permission is scheme an id um, so you should think of this as zookeeper as really weird authentication like you can say um Only a machine with this ip is allowed to do this thing So a scheme here is something like ip or I think they have a bunch of other things like username password I think there's like everyone and then id is sort of the value for that scheme So if the scheme is ip then id would be the ip address here We could do something really fancy like we could have an enum to Sort of expose the different kinds of zookeeper schemes that are supported. I think that is making the api nicer But it's not something we're going to do now But probably something we'll want to do so we'll leave it to do there and then perms is if if you match this then what permissions do you have? um, and you see here permissions This looks on awful lot like uh bit flags And so we may end up Making this in fact, it might already be a bit flags I know they've implemented them to themselves If you look at this right, this is totally a bit flags So we'll probably use bit flags for this um All right, so we're going to need these types Uh, oh, I've got some other things too Oh, I see there's like a convenient method for making Why is this unsafe? Oh, I see the unsafe in the sense of if you set this acl everyone has permissions you just prefer explicit name value Um, so you're totally right that I In fact, you will see that once this code actually compiles. I think my uh rust format will actually simplify this for me Um, the reason I I start doing this is because every now and again I like find that I have to add an into Or I have to like do some other like trivial transformation on it in which case if I just had this I then have to erase it and do data.intu So I like to start out with it just being um Pairs until I know that it's right and then I simplify and usually rust format will do that for me like in this case expected cow found cow Seems unlikely. Oh, yeah, we'll deal with that in a second um All right, so I think we're just gonna I think we're just gonna stick as close as possible to the the rust sukeeper API for Now and then we'll see whether we start Whether we start Modifying it to like use bit flags for example, so I think we'll just like Take this entire file which contains all the acl stuff um I think we're gonna make their source Types and we're gonna get move source types rs to source Uh types mod rs. This is no longer necessary in the new edition, but until we get the path change, we're gonna have to do it Types mod now has these things which We'll probably let stay there for now Um, and then pub use acl star And now acl dot rs All of this stuff We also know that a permission Oh, I see. No, it's fine. Uh permission will probably Should turn into a bit flags, but it's more work to turn it into a bit flags And so therefore I won't do it now. There are a lot of these kind of changes that I want to get all the stuff working first and then we can start building nicer APIs around them Uh So now that should give us acl Once we do this And now the question is wait, why does he complain about this? Cannot find acl. That is not true No cargo check. Did I like misspell acl somewhere here? Oops. Oh, this is Really? I mean, I guess Am I doing something really silly? Oh, yeah, it's this This is also something that will be fixed with uh With the new path changes that if you want to use something Then it can't be from a module unless you explicitly say self like if I do this What it thinks I mean is take a top-level crate called acl and extract this type Which is just not what I meant And that now acl is in scope here and so now we can just use acl because now This use brings acl into scope. So that gives us those things. Oh, I guess we need lazy static Lazy static. I think lacy static is 1.0 I think that's reasonable Right I don't think we should actually need lacy static for this. I think you can uh, I think we could make the Uh new for acl be const Um And that would make this problem go away I also think that for acl specifically We might also want to use cows Because you could totally imagine that the user has some lists of acl acl. So that's um static So I think we actually want a This is a we play the same trick except it's going to be static over acl's And now of course, uh Proto request This is now going to be a cow static acl list of acl like so Uh, it complains about not finding acl. That should no longer be true I think that's not true. Um, all right. How's that looking? Expected cow found type parameter Right. Yeah, so this is the place where we need the intus as we discussed earlier with um using the Initialization Because we say the d and a have to implement into cow because otherwise the user would have to type like cow borrowed or cow owned Which is really annoying But that means that d and a are not of type cow There you can just be converted into it. And so that's why we have to use the into and then we can't use the The when you call it eponymous never heard that word before Learn as long as you live all right That's getting pretty close 81 right. So this is there's no longer a bad version. So the question now is what kind of responses can we get to a uh a create And that this will tell us Uh, keep your exception. Of course invalid acl is the only thing you can get back Which is a little odd Yeah, that's a lie. Okay, so you can get node exists. Why is this throws wrong? So you can get node exists. You can get no node Uh, you can get no children for ephemeral Uh, something about size And something about invalid acl so This means that result will indeed return a result. So we're playing the same trick as with the uh With delete where there are a bunch of different delete specific errors you can get where the There was no protocol error, but some operation you did failed And so we're going to play the same trick here of if it indeed Finish correctly, what does it return the actual path of their created node? So it says actual path because imagine you're creating something that's uh Sequential then you need to know what integers it added to the end Uh, and then this is going to return an error create Uh, which we have yet to find I think realistically we're not actually going to get this So a create can either give you what did I say node exists target node already exists It can give you a no node Which is also a little bit weird um Parent node did not exist in theory. We could uh store here what the parent path is to But I think we'll just not parent node of target Does not exist uh Eponymous having the same name I guess that makes sense Thanks, learn something new No children for ephemerals, which is a Terrible variant name But okay, uh And what this is saying uh parent node is ephemerals and cannot have children Uh, there's something about size. So I guess here we're going to look over at our uh proto error not empty System error, okay, so don't those it doesn't seem like there's a well-defined one for this Invalid acl So that's the other one we can get Invalid acl The given acl is invalid I think one question I have is Is there no way for zookeeper to give more information about the like if the acl is invalid does it not give you an error message? That seems really odd Um, I feel like that's not true I have a very strong suspicion that whatever Whatever is left of the message you get back is the error But sadly, this is not really well documented. Let's see if we can tease it out from here. So, um So in proto Read from Yeah, so empty response does not read anything in this implementation The question is in I guess it's in i o maybe What did they do When they get an error Handle response. Okay, so let's look at handle response This is pretty familiar code at the moment, right? So this is like Make sure you have four bytes because until we have that we can't really do anything And then to just handle chunk Oh bytes.freeze They're doing something funky What is bytes? I feel like bytes is something that we might want to might want to use here somewhere Oh, I've heard of this great, but I've never actually ended up using it Wait, this seems fantastic This might be something we want to use instead of I don't know if you recall But we do this this trick where we keep track of how far through a vector we've read It might be that we can use this instead um But let's put that on hold for a second So if it does handle chunk And we're not in connecting it reads the error The header And that should all be fine Now If the xid It's a closed session. No, it's not that's a client Where does it handle? Right, so a raw response is what we want to know Send response Probably also know what we want So there's a zookeeper raw response Let's see what it does with that So it's a reply header. That's fine. Where does it actually get its errors from? So request Creates a raw request Gets back a raw response match response header error And there's a zk error from because it seems like it doesn't The rust implementation certainly does not read out the The data from an error I find that really hard to believe There's no extra message z error There's definitely a message being written out here in the HTTP logic Question is where does it go? Request header maybe here. So let's see. What is the is there an error response here? Set ACL response Just returns us that error response Really is just an end error Well, I guess there's no additional message does the If you get a keeper exception keeper exception Does it include more info like What's the invalid ACL exception? Is there really no additional information? No, apparently not That seems like bad design, but okay The given ACL is invalid was invalid. I guess nice invalid All right. So those are the kind of errors you can get with a create Um, and so that's the kind of error you can get back So what we will do is now do the same thing as we did for exist We map all the different responses. What is the actual response to a create? I guess is what we need to figure out So let's see fn create So create Is a create response What is the create response? It is a Just contains a path Right, so the thing we read back is really just a string If you look at it, all we do when serializing a response to a create is just we read a string and so this will actually be a string s I'm going to assume that there are also other methods that Return strings. So if we go to our why is there a file called tick that seems wrong in so many ways So if we now go to our response We're going to have this sort of general just like we had empty. We will also have string And I think we already have that set up because that's going to be a string reader So if you get a Create request and it's going to be a string Which is going to be a Reader dot read string perfect So if we get a string, then we're all good if we get something else got non string response to create Which again should be unreachable And now we need to of course map the errors into the different creators. So there's no node, which gives you no node There is not there's not not empty, but there is node exists Which returns node exists There is also a valid acl which gives you an invalid acl And then anything else is a protocol error. Is that right? No children for ephemeral's No children for ephemeral's same thing here Let's see the check In the error, you forgot a comma Nice. Good catch. You are indeed correct, sir. Variant invalid acl not found here Technically String s not found in scope in response 13 19 oh It's a string That's a little awkward, but I think that should work Now in request online 92 Serialized into right. So the question is how do we serialize a create? Well, we have a path We have a mode We have a data I guess we just have reps to all of them Mode is probably copy would be my guess Is acl copy Wait, where is acl? is Why are the tests? That's weird Uh Ah acl is only clone. That's fine, but mode Should be copy. That's not true copy And that is probably also copy actually Um, so we take a mode We take a ref to the acl and we take a I think that's all And the question is how do we serialize it and it's basically we do the same thing as we have done all along We Don't forget that So we write out the opcode of create Uh, could there be no additional information on acl for security reasons? That's a good question. Um Well, so remember it says invalid acl and it sends that response only to the person who tried to create the acl So I'm more thinking things like if you try to use a an acl scheme that does not exist It seems reasonable that you should get an error that that scheme is incorrect Whereas like If you get no information back, you don't know whether you like mistyped an ip address. You don't actually know What the error was? That's a little awkward, right So let's see what do we have here Uh, the path the data the acl and the flags Okay, so writing out the the mode is pretty straightforward because all we're really doing is uh Writing in does the night 32 Writing out the path. We already know how to do just do this Writing out the data. I think is basically the same That's a good question. Actually data here stores a veq u8 How is a veq u8 serialized? write to for veq Is basically the same as writing out a string? I mean they are basically the same Yep, they're the same Um, I don't know whether we have this write to though We have it for slices which is basically the same Which I think means that this implementation for string can really just be writer dot write to self as bytes Should be the same although it's a little bit less sufficient Um, so this one's going to write each byte on its own. This one is going to write, uh This is going to do a single sequential write of all the bytes This one's going to call write to for each byte individually, which is actually really not what we want to do for data So I think what we'll do is uh implement write to for See, I don't think I'm allowed to do this sadly Uh because these two implementations are in conflict. I think with specialization it should sort of be okay But I don't have a way to specify that um The alternative of course would be to not have this generic implementation Here's what we'll do We'll have a generic write list which takes This and returns you an io result And its implementation is I guess it needs to take our writer as well In fact, this could even be generic over w Um, and it is going to do this and it's going to be ts Because now you can always opt into the slow one, but the default will be fast Which I think is more along what we want and see how it likes that Um Because then we'll use write list for writing out the list of acl's for example Where we would need to serialize each element independently anyway, but for things like uh u8 Actually, maybe it is only u8 for even for like an i32 you would have to do endianness But I think it's important enough because data is the thing that wouldn't be the largest if anything And the zookeeper things are generally not very large Um, and so what we'll do is we'll have a fast path for u8s We could flip this around and say that We have a specialized function for writing out u8s Um, and you call the method for non u8s, but given how slow this could be I think we want to do it this way around so you don't accidentally do that Because then here we'll do write list, uh Mute buffer acl This will be something like Because no, this should be just this I think that should be all we needed Uh 139 Fails because try adding a semicolon Or removing the line altogether Missing where t is right to oh did I? We're probably right Yeah Thanks 139 Expected i32 Okay, so mode This is why we put repper i32 on it because it means that this conversion is fine acl does not implement write to that is totally true That's why we need to find the write to implementation Imple write to for acl Great It is so handy to have an existing implementation to start from All right, so implement write to for acl Uh, and it will Herms dot code I find that very hard to believe that that should be necessary permissions Oh, it's because it's fine All right, uh Yep, that all seems fine What else 63? What? It's written. Oh, right. This returns an IO results What else do we have 98? Something something itter This is now a slice so we don't need this anymore Self-valuable is only available. It's not self it is ts 99 wait, why is this written this way? If res is error then return res so this can just be this I think I spent immutable reference That's just write to Yeah, the question is whether we want dynamic dispatch here or not. Uh, probably it doesn't matter terribly much Uh, I think technically the way to do this is This and then we change all of these to do this Where did I do something silly? Oh, all of these need to be that's unfortunate like so Patterns aren't allowed in methods without bodies It's fine Uh, let's get rid of the source error failure business Uh, all right request 151 Right, if you get a create requests Then its opcode is create a 143 Right, all of these now have to So it used to be that took a mutable reference to anything that influenced the right trait It means that the implementations are no longer generic. So they do dynamic dispatch on the So it would do dynamic dispatch on the underlying writer, which is just unnecessary Um, but now that we made a w we don't have to pass in the buffer because that would be Giving away ownership of the buffer. We just want to give in a mutable reference to it So it's going to change this this I think that's all 142 Consider changing this to mute buffer. That's weird Wait, so in theory, I should be able to do this. So why is it not letting me? Because buffer is already a mutable reference Right, so what this method is given is a mutable reference to evacuate Yeah, exactly And so therefore it shouldn't actually take ownership of it when I call Oh, I need to reborrow this is So this is notion of reborrowing which is basically I'm doing This which is a no-off really But it means that I'm creating a new borrow instead of giving away the one that I currently have And so now I think the compiler should be happy with me Uh, it mostly was except for here 99 Just to do this Great 97 A variable does not we don't even need rest anymore Because this just returns an okay nothing Uh, 96 Result that must be used that is true How's that great All right, let's see if all this actually worked out so what we should do now is We know that we ran the test before so in theory foo should not exist And so we will do zk.create Slash foo With some data that's going to be Is here we can test out our new new thing. Hello world And create mode Persistent and What is the I don't like the ACL coming first but ACL Open unsafe So like so and then we want to inspect it Which is going to give a path and we're going to assert equal that path Is equal to I guess this has to be Probably ref path Uh, we're going to assert that the path is equal to slash foo and then Is zk Then we're going to check that it exists and we're going to assert Not equals that None right, so we're Because we created it. We know it should not be empty In fact, we can make an even better assertion here Now we can actually write it as a test right because now we know that it'll pass or should pass every time Uh, we can do even better here and say stat dot unwrap Which basically so unwrap is basically an assertion that it is some Uh unwrap dot data length Uh Is equal to A low world dot length and then we delete it And here we just assert we just assert true. Well, I guess here what we really want to do is uh We know there shouldn't be an error, so we don't even need to inspect here Because the error if this if this delete produce an error Then we would get and hit this unwrap because the operation actually failed And then exists is going to do Basically the opposite because we now delete it right so we should assert That stat is none if we do this Uh, and then We're gonna map out the zk perfect. Let's see how that fares Not so much can't compare Oh, right because This path is technically Uh so These remember that delete there's delete error and debug errors. Sorry. There's delete and creation errors and those we actually do have to Uh compare partial and for this So we actually want to check that the The create there was not a protocol error and also the creation actually succeeded And so that will be the same for Delete we do actually want to inspect it specifically We want to assert eq res okay Right because there could be that there's a deletion error and we're not expecting a deletion error Uh in theory, I guess we could map here, but let's do it that way first So we should make a super fancy assert macro so you can type assert equals equals Yeah, it would be pretty nice. I agree. There's also a really nice thing called, uh Oh to call rust assert deep equals pretty Like pretty asserts or something Uh, I think it's this one So pretty assertions gives you full diffs of structured output Really quite nice So that's something to to keep in mind Uh 155 I messed up didn't I where did I mess up? Really? Oh And somewhere else and somewhere else In fact, I messed this up even worse than I thought like This There we go. Uh can't compare Can't compare result string create with result stir To well, I guess we could I could technically map this into string aster perceptive signature found signature mismatch type on 154 expected i32 found u-size as u-size There's a lisp stream now. I don't think you'd even need a proc macro for that. Oh, you're meaning you mean for, um Oh for doing assert double equals Yeah, the double equals inside assert. I think you can in theory do You would have to parse the token stream I guess you could just if you if you can't figure out that it's a well recognized pattern Then you just don't rewrite it. I guess it's the way to get around it The problem with the language name Oh, yeah, there's a lot of rust is hard to search for the trick is to search for rust lang Similar to for go you do go lang And that way you get basically the same results because usually the search engines are smart enough to do with this Um Convert from Oh, that's interesting A reference to a vec does not implement into cow that seems wrong Specifically, it seems like there's no implementation in the standard library for Imple Imple t Into cow t four Is this Or take a vec t This should definitely be in the standard library, but apparently is not because just should them be cow borrowed Uh, this implementation should be in the standard library and is not that's uh That's pretty unfortunate because it means that we can't so opal open unsafe returns a vector although it doesn't have to anymore Uh, where's our acl? So where's our lazy static? This No longer needs to be that this can now be a Uh, acl I think In fact, I'm fairly sure This can be const fn if Oh, no, because it makes strings Uh, it's awkward Yeah, I think in theory you can do this And then this would return a in fact Even if we kept it the way it was But um, we should be able to do this Yeah Watching my writing like so Like so Could be lazy static complaints about that though Really? Oh right, uh one Found type acl open unsafe Or really this maybe Makes me a little sad that that has to be uh Because the acl also doesn't need to contain strings. It could contain cows But then I guess a little bit annoying But that seemed to have worked. Uh, let's go back to our lib Uh, thanks. I appreciate it Um 159 It's saying that it can't compare a stat because stats aren't comparable Ooh That's a good question I think stats should be comparable There's no reason for us to not do that So let's do partial ek and ek Uh As well Now it's saying it can't compare Oh because one is a reference and the other is not So we need to do this Uh 157 Can't compare unless the same issue 144 From that Oh That's really sad So the issue is uh This is actually an array of length 11. It is not a static slice. So we do need to do this It's still less to write than cow borrowed Or vek from Uh Oh the acl has some doc tests somewhere Uh, let's get rid of that And is there any here or no? Let's try that again Well, that seems to have worked So there's the create There's the exists There's the delete And there's the last exist And the unwrapped did not fail Let me just It's pretty neat if that just worked on the first try, but Yeah, like it's a zookeeper, which we are fine dropping Just so that I don't get confused by this later Um, well, that seems to have worked actually So now we can create we can check exists we can delete we can check exists again Uh, and we have all the create modes And it seems like the length was right too because otherwise that assert would have failed So I think we actually now have all of those basic implementations that we wanted Right we have create exist and delete Uh, I think then it's time for us to move to watchers Because that's where things start to get a little bit hairy There are some other things that we have not yet implemented that we will need to In particular, uh, in particular So remember how zookeeper have these, um, these ephemeral nodes Oh, right commit. Yeah, sure, sure, sure, sure, that's a good point Create plus delete I normally write, um, better commit messages, but On a stream it's not worth it So many keys Um It's been pushed It is alive, right? Isn't that nice? We have a fully self-contained test Um, yeah, so remember how there Ephemeral nodes in zookeeper So ephemeral nodes, the idea is that once the client goes away, the node is automatically deleted by zookeeper But this relies on zookeeper being able to know that the client has not gone away And you can imagine that you have a client that like creates some node And then it doesn't really do anything for a while And so zookeeper actually requires that we do heartbeats And that is not something we've currently implemented and it's going to have to go Somewhere around here Specifically in Actually, I don't go here So this is going to be Actually, I think it's going to be down here If it returns not ready here To do send heartbeat Actually, it's not even here Because this implies that Yeah, it's actually not even here I think we're just going to have to hear at another poll heartbeat And so we need to make sure that we're going to have to spin up some kind of tokyo timer And then whenever the timer expires, we have to make sure that we send a heartbeat And we can also reset the the timer whenever we do an actual request Because then zookeeper knows that we are still alive I think this is a pretty low priority thing We can decide whether we want to do heartbeats first, which is sort of slightly lower level So it goes back to some of the stuff that we wrote In part one Or whether we want to do watchers first and then come back to heartbeats later I'm fine doing either So Sort of up to what you want to see first I think we'll probably have time to do both It's only two I will go pee though And then I'll be back I vote for heartbeat I vote for watchers This is the first time So you're excited Great So we have one vote for each and one person excited All right, let's do Let's do heartbeats first because otherwise I'm going to forget to come back to it Because it seems so trivial But it's something that just like if you don't have that your client is just wrong Whereas watchers is a feature So let's do it quickly And then I'll have to make it up to you by getting to watchers quickly All right That's all right We're just trying to decide which feature to do next Or whether to fix a bug or whether to add a feature I think is really the way to phrase it All right, in theory, this should be very straightforward So the idea is that we want to make sure that at least at every so and so interval we Send basically a ping to the server like a heartbeat request to let it know that we're still alive We have this this packetizer which is the core The core future that we're polling And basically the way we're going to implement this is that Future is also going to contain a timer future that we're also going to pull every time we go through And if that future is expired, then we know we have to send a heartbeat So A packetizer we're yet So that's the stream and then we're also going to have a heartbeat timer Tokyo has a built-in timer module In this case what we want is to It basically re-exports I guess Technically we should use the documentation from here I think the goal is eventually for tokyo to move to Not re-including everything here. So have you include all the all the crates separately So that they don't have to do a tokyo release in order to release changes to the sub crates But for now let's deal with this What we want is interval And I think if I remember correctly that interval also has a reset Oh, it is not just a new I guess we could do delay instead Yeah, let's do delay Delay great Um, and so when we create the packetizer Somewhere down here So we're gonna do the connect And where's our So poll in queue is setting up in queues Right is when we write stuff out Read is basically where we want to see Down here That's the part where we get a Connect so remember initially so uh, we're going back to here somewhere Connect response So a connect response contains the timeout and the timeout is what tells us How long to wait until we issue the next heartbeat after every request So if we look here in the zookeeper docs Somewhere around here, uh nodes and ephemeral nodes Uh, it's not here maybe zookeeper sessions So they have this terrible terrible diagram, but the basic idea is when you connect, uh, then I doesn't even say in here. Where is the The Here yeah, so session is man session exploration is managed by the cluster It provides a timeout that is used to determine when the end expires So we need to make sure that we get back to it within that heartbeat So this is another thing that we're actually missing that could be interesting to implement is that Zookeeper has this notion of if you get disconnected from a server Either because your session expired or because the server went away The library has enough state to re-establish a new connection to a different server or to the same server And continue with the same consistency model that it had by sending along some special values And that re-establishment of connections is not something we've implemented yet But it's totally something that we should implement and in fact the basic library does Um And so that is something that we will have to add at some point It could be that we add this as a part of uh Of heartbeats, but we can probably deal with them separately Uh Thanks. Have fun. Thanks for coming Let's see But so the idea is we don't actually know the timeout until we hear We get the first connection response And so what we do want to do here is actually um If let response connection Or connect Yeah connect We want to sort of fish out the timeout here So if we got a connect response, then we know that we now need to set this delay So now we can do self dot timer Is equal to or dot reset to it's sort of like time Instant now plus timeout. It doesn't actually say what the timeout is in Which is a little awkward. I guess we search for timeout. That's fine Reads an i32 that's fine Oh, that's unhelpful I assume that it's in seconds Uh But it doesn't actually it might be milliseconds New connect request Okay, so the question is what does it send in its connect requests? So the connect requests Well, that's sort of unhelpful con resp Where is con resp initial timeout milliseconds sounds a lot like it's in milliseconds um Connect response So this is a response. So the connection request The timeout is zero. That means it expired. So we haven't dealt with session expiration yet. This is one of the cases where we need to reconnect but generally What is this Two-thirds of the timeout. I think I read somewhere that it's already set pretty conservatively um But I may be misremembering Uh this ping not only aroused it to do the timing of the ping is conservative enough To ensure reasonable time to detect a dead connection So the question is like how close to the heartbeat ratio do we want to be? I think again, we'll stick close to this implementation just because it's one that we know works um This is this like you can be in read only mode, which is not something we currently support Um, but it's also something we'll have to deal with But it certainly seems like the value is in milliseconds So it's going to be uh time duration From milliseconds, which is now the standard library, which makes me very happy Um Oh, can I that's a good question. We'll need to know what the uh What the timeout actually is so we'll have to store that somewhere too Uh timeout is going to be a time duration So the here we're going to do self dot timeout is time Duration from milliseconds timeout and it's going to be two times that divided by three Um, and then this is going to be self dot timeout And then we're going to forward the responses we used to And now we're going to change down here is I think we're going to have to So the tricky part here is actually that Whenever you do a write to the server we sort of want to reset this timer, right because uh This is the server needs to get a heart or a ping from us if it doesn't hear from us otherwise What this means is if we're already writing stuff the server we don't have to send pins And so the way we'll do this is every time Uh, we successfully write out we can reset this So I guess wrote is false Wrote is true If wrote remember that pole write gets called every time even if there's nothing in our buffers Oh So if we wrote something up out then we reset the the timer for I guess we could call this heartbeat timer And then what we want to do is Uh where we call all the pole methods, which is down here We're going to do polling queue poll read poll write and then we will do self dot pole Timer, which is really just timer dot pole Uh And what we're going to do is we're going to match on it Yeah, uh, we're going to match on the timer Uh Write is wrote nice That's why I want the rainbow colored variable names that there's like a plug-in that does this which is Sometimes very handy and sometimes just really annoying too many colors on screen We actually have a little bit of a challenge here because okay, let's say that this comes back with async ready So async ready means that the timer is expired and we need to do some stuff If this comes back that way Then what we really need to do is we need to queue some stuff to be written So what do we do in poll in queue again? We Just add to the outbox Right So we'll do self dot outbox dot extend and then we'll write whatever the byte value for Whatever the byte value for a ping is we'll probably construct it and whatnot. This is more I'm just walking through what we'll do. Um, however I guess we'll do question mark However, we then need to poll right again because otherwise We don't know that the underlying stream or we wouldn't be notified again. So imagine that We go through these and there's nothing to read or write And then the poll expires so we extend the outbox And then we return from the the future if we return from poll But there's nothing that causes us to be notified again. So that we actually write out these Bites that we include here. So we actually have to do poll write again um Which is a little odd I think This is why I sort of want to do it before poll write, but then we have a race with the timer Actually, let's do that It's a little bit easier to think about And I guess we don't really care if we get not ready And let this equal to that then Right, so the idea is We check whether the timer is expired and if it does we we add a ping to the outbox and then we poll write We have to poll write regardless because there may be other things in the outbox or if there's nothing else in the outbox I guess here we could even if self dot outbox Dot is empty We only need to add to it if it's currently empty because if it's not currently empty They're going to be right. So we're uh bites that we're going to write out regardless There's no need to add a ping because there's already a request in flight. We can do that here Request in flight. So no need to also send heartbeat Um, the other thing we'll have to do of course is we'll have to reset the timer Because we want to We want to keep sending heartbeats, right? Let's see. So the question now is how do we send a ping? My guess is that pings are pretty straightforward. Um, if you look at our poll in queue What we can basically do is write out a very simplified version of this code Um, it is simplified because all we really need to do is Because we know the length of it. So if you look at a ping Uh, there is no other ping in this file So it's an i o then I guess yeah So here you see he's just hard coded what that looks like and it's sending an x id of minus two minus two It's quite odd But sure. Oh, it's because he just has it's so that there's some id that you can scan for that you can ignore I assume he does the same for connect that connect is like minus one or something Um, but it well the reason to do this is When we get a response here if the x id is minus two, we know we can just ignore it So here, uh, if self dot x id equals minus two it is a response to heartbeat And so we continue I guess Is that right? Right And so really and we also know the length of this So the length of this will just be the length of a header Which if you recall Is just eight bytes, right? So the header is just the x id in the opcode the x id is an i32 the opcode is an i32 So we'll do so self outbox Right minus two And then the type it's going to be outbox Right, so if we remember from Sending out requests the thing we do always always write out the i32 of the opcode first So in this case, we will do the same thing. We will do self outbox write i32 and this is going to be opcode ping So we write this should not fail opcode And there's basically nothing else to write out. However, we do have to write out the length and the length is known for pings Specifically it is This is an i32. So this is four bytes. This is another i32. So this is another four bytes. So eight bytes total This can go away This can go away That can go So this is Send a ping Right, does that make sense? So every Every time the timer expires we reset it and we also send a ping if there's nothing else to write In theory, this shouldn't affect our implementation at all. Oh I didn't like that at all Quest opcode is that right quest Missing timeout and timer. Oh, that's right when we start it We don't know what the time timeout is. We also don't know what the timer is Um, so this is going to be time duration Uh, something like new Zero And the timer is going to be a tokyo timer Delay new and it's going to be something like time instant Now plus actually that's a good question. Is there an instant like forever into the future? I doubt but Yeah, so it's now plus time duration From What kind of durations do we have now? from I guess from sex And we will make it 86 400 which is one day Right My missing messing up my uh, 24 times 60 times 60 Yep So we're basically setting a timer that will not expire We could set it way harder higher if we wanted to and then we're setting a timeout to zero because because the timer will never expire The timeout will never be used until we set it when we get the response to the connect Unused module time that is entirely true. We need time And expected two parameters It's fine and 259 Function or so did it item not find in duration. Oh millies Millies U64 found i32 Sure U64 from Really? Oh because it can be negative We happen to know that it's not negative. I think Timeout is greater than or equal to zero as U64 Failed That's interesting uh Unwrap none somewhere at proto mod 250 Oh not self xid xid Surprised that even worked Why is there even oh next xid to issue? Yeah, that makes no sense. Okay Well, that seems to be stuck in some kind of endless loop Length is 16 for a very long time apparently Why is the length 16? That's a Pretty long way back to go That's that's a lot of lines ha God request create What is this length is 16? Where do we print that? Hmm. Well, that's kind of interesting Handling response to create session read more bytes So that's into the timer Huh That's a very good question So this suggests that we're not getting a full response to something Hey Why are we not getting a full response it's a very good question Length is 16. Let's see what those 16 bytes are shall we? Any print line? Oh, this is gonna it's gonna spam my terminal so much Self inbox self in start plus four to self in start plus need 255 255 255 254 That doesn't look right at all And I think what's the first? X ID is the first thing we get back. So my guess is that is an X ID of minus two So E print line Oh Yeah, yeah, yeah. No, you're totally right. It's the continue in the pink case Uh, because it doesn't get to continue Without also Incrementing in start What is this loop? Is there a loop? Oh Yeah, um, okay, first of all this Should go here Second of all No, I think that's right And then we do we do an else instead to here because I also want this to get executed if we get an X ID response response to To user requests Good catch perfect Pactizer exiting reactor gone. That seems all fine Beautiful. All right. So in theory, we now have Pings So the way we can observe whether this is in fact the case is we go down to our test And then here we do something like timer Oh thread Red Sleep ms. Oh, I guess we don't actually know what the timeout is It's a little awkward. I should probably have figured out that out Oh, wait. No, we set one when we connect, right? What did we set it to timeout zeros? We let the server choose So if I like sleep for Uh duration from sex like 10 seconds Is I should probably print out what the true timeout is Um, where is it? We set it here Timeout is this many ms So the timeout is how much Timeout is 4 000 milliseconds. Okay, so in theory. Yeah, so here we're seeing pings, right? So this is after the test is finished and that's a ping 2 3 I guess I'm about it counting finish 1 2 3 That seems like about four seconds great All right, so those are in fact heartbeats Beautiful So we now have heartbeats Good job team I get diff I guess I don't actually want to leave the sleep in there But apart from that, we are all good Send heartbeats Oh, thanks. Uh, yeah, the dot files are in github over here Um And in particular of mrc. So this is the um From base. I'm using the base 16 themes and in particular. I'm using the one called atelier dune So this team that's pretty nice To oh, sorry, you're totally right. Yeah, it's two thirds or four seconds. So that's why it's not actually four So maybe my counting to four is actually exactly precise I'm gonna I'm gonna go with that even though it's probably not true Um, okay, so watchers So much signing Okay, so watchers are a little bit finicky, um, they're They're pretty straightforward in the sense that they're just a way for you to get notified when something happens Um, so the basic idea of it is something like In fact, maybe we should implement this with gets first given that we don't have get yet Get yeah, that's funny So there's get data Right, so if you notice the signature of get data in the original library, um If you ignore the watch for a second get data Takes a take self takes a path and returns you a stat of that thing and it's the data it contains So far so good. This is watch parameter two Uh, what watch does it it says whether If this changes you should be notified and the way you're notified is through a watcher The easiest way to understand this is to first look at the get data w method where you pass in a watcher And a watcher is basically just anything that implements something that can handle a watched event and watched events are things like A node went away a node was added data change those kind of things Uh, if you use this method and just say watch true, then the way that works, um The watch it doesn't actually say here, but you can add a listener And that listener is the thing that will be notified Somewhere up here yeah, so When you connect you set a watcher and that watcher is the one that will be notified if you pass true So you don't pass an explicit watcher. You just pass true. So this is what's known as the default watcher for zookeeper So the default watcher isn't always notified of things like You got disconnected you got reconnected those kind of events But we'll also get any of the other events if the such things occur Um In the asian world we have the advantage that we can use streams for this The question is just when will we return the stream? I think the idea is going to be that when you connect You'll get back a zookeeper and you'll get back a stream That will be the default watcher This is sort of dictated by zookeeper that you have to have one because you need to notify them when A connection events or disconnection event happen Um, we could alternatively have zookeeper itself implement stream Uh But then it couldn't be clone And I think we want it to be clone Hmm The basic idea is this is going to be uh, this is going to be a futures mpsc If I could this is a single channel, but there's no such thing in futures at the moment um And so the receiver isn't clone so if we stick one in zookeeper zookeeper wouldn't be clone Um So that's why i'm thinking we might just want to return Two of two things like when you connect you get back the default watcher and zookeeper And then the zookeeper remains clone because that's the thing you use to To execute additional methods, but the watcher there's only one of and you can't clone it Yeah, I think that's the thing to do So connect will now resolve into both a self And a I don't know if I'm allowed to put stream here Don't actually remember So the item is going to be a What's this watched event? Again, we'll stick to trying to use mostly The types from From here in fact one thing we could do is exactly reuse the types But I don't think there's an advantage to Oh, this has so much stuff in it Yeah, I guess we'll let this stick around So this watch type this keeper state This keeper state So it's presumably further down Well Let's stick with this and then we can we can simplify if we want so this will return a watched event And the error on that will be In fact, it will never error, but let's just say that it's failure for now I think this will just never occur, but I don't think we have a way of saying that Um Right, and then we'll do something like Types watch rs that contains all this stuff And then there's a keeper state Need to find Consts Oh, what's watch type? I don't actually need watch type. Do I Why are those defined in a different file? That seems just odd All right, so they have a source consts Which includes then Keeper state and watched event type and these are All just straightforward. My guess is they're all i32s, but let's double check So Keeper state That's gonna be unfortunate Uh watcher event State of the keeper client runtime So this is definitely a repper i32 I don't think we have the So that's an i32 it is also copy um There's maybe a path and there's a type and that type is also an int So this is also repper i32 It is also copy and we don't have this even though maybe we should So Now when we have a lib we're gonna out of here use watch event Types watch no types mod It's also going to bring in mod Watch and hub use self Watch star And I think we'll also have to at least for the time being This will include watch the when event keeper state And watched event type now way we can return that in the stream um Which means this is now going to return a Oh This I think russ might yell at me here. I don't think you're allowed to use Uh infiltrate inside of an infiltrate, but I could be wrong Uh in fact, we don't even need to Uh, we could just create the stream out here So I mean because we know what this is going to be It's just going to be a future sink and psc The uh Unbounded probably um And then this is just going to need the uh The transmit end And that way we can map this so when we get back the um the zookeeper connection We map it into a zk and an rx Right because now this Uh takes the tx For the watcher It's default watcher Uh, and that is a future sink mpsc unbounded sender of Watched event And now packetizer new is given that default watcher And now proto mod Down here somewhere So this is going to be given that same default watcher Down here, um, we're going to have to stick that somewhere So that we can send on it whenever there's some watched event that occurs Watcher to send Watch events to pretty terrible documentation, but it is true This is going to have to use Watched event This now just takes the default watcher, which is great And now of course the question is whenever we get a watch to when we need to make sure that we send it The way that zookeeper communicates Watched events we can take some inspiration here from source io So if you look at it here Somewhere down here, let's see if it's readable Handle response Uh handle the chunk Yeah, so xid minus one Is an event I think if I remember correctly Uh, I guess we can search for minus one and see if it pops out anywhere else Yeah, I don't think so Uh, so the idea is that because any response you get that is, um That is a watched event doesn't really have a An associated request with it right it could but it does not necessarily have one because It could be just like you got reconnected or the server is now in read only mode or any any such thing So what we want to do is Down in poll read so remember we had this x equals two for the heartbeat we're gonna have if it's minus one Then We're gonna we got a watch event And now the question of course is Here it seems like it just like wraps up the raw response, but I think we want to parse it out here probably um So what is watch message? Watch watch message I just want to see what the I guess we could also check the jute file It seems like watcher event does not that's curious So if you look at this, um It still has a header Which means that it has an xid and an opcode Ah, no, right. So this is a response header So response headers have xid the zookeeper id that is used to keep track of How many is essentially a way to guarantee consistency? We haven't really seen it in play yet It'll come back a little bit later potentially when you would you reconnects Um and an error code and in our case The minus one is the xid and we can ignore all the other values The question is how do we parse what follows and we parse what follows as a What was it? Watch event Right, so it's just one of these things Um So the question is what do we do here and I think the answer is simply that we uh We just parse out a watcher event Which we could do in a couple of different ways I think what I actually want to do here is Imple read from for Right, because um We have this now watched event thing and we want to construct one of these essentially by reading out What's in what's on the wire and we have the the setup from here specifically we know that we Here Uh, we know that we first read out the type So the type is going to be a read dot read by 32 We know that all of these things are big end in I guess I can't use type. It's going to be w type for watch type Uh, we also have a state Which is the State of the keeper client runtime Don't really know what that translates to I guess it's uh, I guess we could look at this here Oh, this is what does the server think that the state of the client is So for example, it might know that we have been disconnected Um, and then most the last thing was a U string And a U string is the same as we do it use for paths. And so that is really just a uh A length and then bytes So we will just do read buffer. I think That is I guess if do we have a read string we do right? Yeah read string And then we can return a watched event because the The watched event remember has an event that has these fields so And now we have all the things that we need Um, we have a watched event type, which is the w type as this We have a keeper state, which is the state as this because remember the repper for watched event type is Uh i32 and same for state as i32 and this is Oh, that's a good question How do we know whether there's anything there? Um, I guess we need to find out how it parses a watch message So A watch message is an event and it gets parsed looks at the error If there's no error This is a watched event read from so that's in proto I've seen proto before Yes, this is basically the code that we wrote right? Huh, it looks like here. It's always some That doesn't seem right. Why would it always be some? Cut ch root. That seems very odd. What does this do? Oh right, so In zookeeper when you connect to zookeeper Initially you can set the starting path you want to operate So that path essentially gets prepended to every request you send to the server Um, and what this is doing is if you get an event back You get the full path from zookeeper and this cuts off that from the beginning So that your response is sort of consistent with you having connected at that path Um, I don't think we particularly care about that. I do really wonder why Or how Path can ever be none Given the code is written here. It's always some which seems weird Uh, it's a good question find watches What does the protocol say doesn't really say anything And it's a little awkward and there's no keeper exception for this that helps as much um Hmm Well, this would be none for session related triggers Oh, so they probably manually construct these in certain places That would be my guess. So like if you Uh, if you get disconnected for example, then the library will just construct A watched event for you. Uh, apparently not there proto maybe No, that's just that Const we know it's not io it could be Yeah, so here for example notice that, uh There's like a we need to remember to send a watched event If we got disconnected and then reconnected but it seems like that's not actually used here everywhere Is it could be that we just make this Yeah, let's just have this not be an option for now and just have it really just be path Those will go away and then this Uh, we will make A comment for now because currently it really is always just a string. Oh, did I not? Oh in response, right So this is going to be keeper state and Watch event type Oh, that's awkward Am I not allowed to cast these even though I know the wrapper is the same? That's really annoying So I know that there are crates to provide this. So this is what the One of the derives we got right up in the original code basically did for you Uh, there's rush really not have this so repper rust repper Auto that's not helpful 32 for example and cast is what I want Okay That's mostly unhelpful Um, it could be the nomicon that has something So the nomicon is really a really handy thing for, um Well, that's basically the same It's really handy for when you need to do unsafe stuff, but apparently Other reppers repper uh-huh Fuelless enum without a repper is still native. I mean repper goes a bit. Yeah But apparently it doesn't give us conversions. This sounds like something someone should submit a pr to rust to like deal with this Um, because what we're gonna have to do is impulse from i32 for keeper state From To I guess this can be code this should be Because we're gonna have to do it unsafe. We're gonna have to do Uh, code as keeper state I don't even know if I'm allowed to do that But keeper state is an i32 Hmm I really don't want to enumerate all the possible codes I guess okay, so I guess the reason they can't implement this from is because it's not technically true Then you could give it a value like so for keeper state if you gave it the value seven There's no enum variant that corresponds to that So that's got to be why they do it. All right, fine So we'll have to match on the code and we will basically take all of these Put them down here move all the comments And then we will do Qa Actually, no, I will yeah, sure this this this Sorry, I'm not doing some of in magic Why doesn't it let me do that? That should be a totally legal macro Okay, so what I'm doing is recording a vim macro, which are really handy for these kind of mechanical conversions Uh, but for some reason It is not wanting to let me do that There we go, and I guess these are all Also, that did not do what I wanted it to do did not stick with the minus. This is minus 112 And this is going to be self self How about now? Oh, I guess an underscore is unreachable unknown keeper state This whatever the code is And I wanted to print it in hex variant sassel authenticated not found Oh, is this can you not name variants through self? Oh, there's an rfc, but it hasn't been done yet. That's why So I think I have to do this Yep And now we're gonna have to do the same for watcher event type I can copy most of this Is rust hard to learn? Uh, that is a good question I would say that If rust is your first language It is probably not that hard If it's your second language, it is probably much harder Because there are a bunch of things you sort of have to forget from the stuff you did before Specifically things you think are okay that rust will not let you do Um, and so in that sense it is harder But I don't but I think I think uh, you end up spending more time Fighting with the compiler in rust whereas in other languages You're going to run your code and then have it be wrong and then have to debug it And so it's a very different way of developing and that's why I think it feels very difficult for People who are new to the language Uh, sorry for people who have experienced in other languages and are new to the language And then type code and then we're going to have to take all of this Delete these, delete these Ah, that's not going to work These are all different Read this This I guess this has to be uh Watched event type Uh, I want to learn rust to help developing a game Well, it depends whether you have any experience with programming from before I think learning to program is going to be is sort of a challenge No matter what language you choose I think rust is probably not the most beginner friendly language, but I also don't think it's very Hostile to new developers you have to think about things like types Which you don't have to quite as much in in languages like JavaScript or Python So in that sense those are easier to learn But It sort of depends what you're trying to use it for so the second question here is For what development purposes would you use rust and I think For me at this point I use rust for everything There's something to be said for use the right language for the job I have generally found that for the kind of things that I do rust works really well In part because it gives me I often want low level control for performance and I don't really Worry about having that low level control, but in part that's because I'm so comfortable with the primitives I do think that for It is probably still not quite there for things like If you just want to code something up quickly, I don't know if the rest is the right tool for you But at the same time it saves you a lot in debugging. I think the language Um Still is quite verbose for things like async programming, but that is something that's changing with with async await Um, there's still some things that don't perform quite as well But it's a lot rarer now because lvm is the same optimizer that that is used for ccode op Yeah, I don't know. I think there are some very specialized use cases like parsers where hasco is really good A compiler development where camel is really good websites where dynamic languages are Often considered very good. Although I think there's some advantage to static typing there as well So I think in general like russ is a pretty has pretty Broad applicability actually But it is I think a bit more of a pain to learn than other languages In part because of this notion of ownership, which means you sort of need to You need to understand at least a little bit of the underlying memory model. This gives us a watcher Which we ignore now This need to use future self. Oh, I guess this can now use Where's our mpsc channel? This All right Let's see how that works So we currently send nothing on the default watcher. That's fine because we're not actually getting any watched events But at least now in theory we have the watcher I guess we don't send anything on it yet because we don't parse out what the watched event is But we should be able to do that now, right? So we should be able to take watched event What's the implementation we have for it? Response So read from is what we want From so we're going to use self I guess use request Read from And we're going to read from the buffer that we have And then we will do self dot default watcher. So here Maybe send to the right to Non-default watchers. So if the user has registered additional watchers, we want to make sure to send to those instead Uh Rust 2018 additional will be an improvement for learning. I think that's true Um, especially things like changes to the module system Uh, non-lexical lifetimes being on by default. I think it deals with a lot of those Issues that you will run into very quickly and just makes them easier to deal with I think that's probably true Unbounded send E Dot And we want to ignore the error here Error because the user may not care about this So if the user drops the watcher, we don't really care We still want to try to send it if it fails. Then that's that's still fine Uh Use proto response read from Isn't that what I did here Oh, yeah, right Like things like this will be fixed in the 2018 edition that you don't have to put self everywhere and this is No read from in Oh response Uh types different usability Read from buff I think that's fine. We just do this great So now we have the stuff that's gonna Tell people about watched events except there are currently no watched events Um, so let's do a diff and dot and then uh send receive and send watched events and notify about watched events So now we have the basic mechanism mechanism for dealing with these watch events And then I think the next thing we want to do of course is uh, have the user be able to issue a request and say that they want to watch it right, so There are a couple of these in particularly what we could do is We could add this for exists We could say watch I'm not sure that's the nicest api, but I guess we can deal with it. Um So watch is That's a good question So exist takes a watch which I assume is zero or one Check that uh exist request String and bool requests. Yep indeed So this is gonna be If watch then one l zero All right, so this is a pretty straightforward change to exist just saying that uh Whether we want to set a watch or not and the idea is that if you set a watch all that really does is um Say that if the condition for this exists changes Then you will send on the default watcher Um, apart from that the api will not change and so the way we'll do deal with this is this will be false This will be true In fact, we could even do We could even do true And true and what we should see is that the watcher should get two events it should get uh one event for It should get one event for this exist suddenly turning Uh Falls once the delete happens and then one for this exists that will become true Once the or that it does no longer exists when the delete happens Actually, no, this won't trigger this will be false, but we can do uh zk.exists first Up here It all exists So we'll do one of these first And assert that it does not exist, but but set a watch And then we'll do a zk Ignore the stat To do this like so Right, so first we check whether exists and it should not and we set a watcher saying if it ever does exist Let us know then we create it and this should cause a send on the watcher, but we haven't checked that watcher yet Then we check exists after it has been created And tell it to set a watcher then when it gets deleted it goes away so that watchers triggered again Um, and then we check exists, but this has nothing to watch I guess we could set true, but we would never be notified Um, and so now We then do I guess Yeah zk So here what we're going to do is we're going to basically assert that we got the the same number of things that we expect on our Watcher wise drop called on zk. Yeah, so the reason it's called here is because the reactor would not become idle so, uh If the zk is open that means that we could still send more zookeeper requests And therefore the the packetizer won't shut down because it knows that it could receive more requests So it's only once the last handle that the packetizer is dropped that it exits and becomes idle And so this shutdown on idle if we reverse these or in that case got rid of this This would never finish because the packetizer would never finish because it's Because it it isn't idle and therefore it doesn't shut down. So then that's why we need to drop it. I guess, uh make packetizer idle Just to make that clear Here what we want to do is we want to take w and we sort of want to we want to receive twice Great. I'm glad. Um, there are a couple of ways that we can do this. Um, futures zero dot one Um, if we find where stream So there are a lot of methods on stream. Um We could just collect it the problem is it won't be done. So I think what we want to do is Hmm, that's a good question. What's the best way to check that this is the right thing I think what we want to do is call Uh, yeah, here's what we want to do. We're actually going to take out the we're going to get the watcher back as well And then down here when we drop the zk and then shut down on idle We're still here going to do uh We're then going to do w dot weight Dot count and we're going to assert That there are no more events at the end right During we know that in fact we could move this further up if we wanted to So maybe that's what we should do Well, I should be a little bit tricky But yeah, so so down here what we want to do is sort of Uh into future I guess map X X So we carry zk along and then here we're going to get something like zk And uh into future gives us a stream future, which is just a future that gives both the item from the stream and the s um, so in this case this will be event and Uh and the watcher that we get back And we here want to assert equals Uh event Some I don't remember what the type is for watched event, but this is basically to check that we got the right watched event Uh I guess we can do a map Uh, and then we give back zk and w Actually, we could do here and then again because we want to then take the uh Read a second element because we know we should get a uh an event from the first Right, so when we initially We call exist and we watch so we get one event when we created and then we get one event event here Because we watch and then delete so we should get a second event down here Uh and that event Is going to be Also a watch event of some kind and then we will just do this will just be a map We'll map this back to zk w so that we can extract the watcher I had to run out for an hour. Is it sentient yet? Almost almost we're almost there And then it can uh 175 Oh into future gives a stream error, doesn't it? Uh, that's annoying. I think we're just going to ignore that because it's a test I'm just going to say that All right, because so the reason this gets complicated is because uh If you call into future on a stream what it gives you back is a future there where it either resolves into the Item that was yielded and the stream and the error is the error that was yielded and the stream So that you can keep operating on it and so here we want to map out the error We want to ignore the stream if an error happens and instead just give out the error That's the same thing we'll do here. Uh, let's make this an inspect Because we only really care about the event and I guess we could do this here And then map it to uh, this is actually given zk No method map error for That thing it's because this has to be here Yeah, it's a very dotted code. It's true. Um, it's really annoying because um The testing future space code so often turns into this Because you have to like do this then do this then do this then do this This is where like having monadic syntax would be really nice Like if you could do use do notation here or something and that is basically what we're going to get with uh async and await Right, then this would be something like uh, let stat is await this Right, we could write this as synchronous code and then this would all look so much nicer um So with the wait this will become better, but for now we're gonna have to deal with this sadly uh 175 Expected empty found failure error I feel like that's probably not true Oh Right actually Right This has to be a failure error Streamer We're gonna be three events you said in the watcher that's going to yield twice and you're also setting the same watch a second time uh, so uh, so Watchers in um in zookeeper are triggered only once So here for example, uh This watcher will not trigger again if food changes again Watchers are always single shot in zookeeper and so This create will trigger this watch Uh This sorry this delete will trigger this watch and nothing will trigger this watch Therefore, there should only be two Spect exists, but Uh But but but but but but 178 Oh Zero oh right zero And then into future. Yeah, I know you end up with futures. You end up with a lot of this like sci-fi like speak. It's pretty great Uh, this is not okay because It requires the output is an error expected result failure It shouldn't matter, but Something is fishy This um writing like really combinator combinator heavy futures code is not nice It's basically complaining that I'm Sort of no longer a future here for whatever reason Oh It's because no it's still that Although I guess technically this could now just be this But it's complaining that This does not give me a result So this futures Item Is not a result that is true This futures item is ooh is that because I misread this somehow No, it's item is an option item and the stream itself Right, so this is a like an event and w Which is all I'm mapping it to Right Those are the same So what is it complaining about? Weird And then it's complaining that I can't use inspect because whatever comes back from this is not a future Which is just not true Because stream future which is what I get from into future So the w we get from up there For whatever reason It's complaining about the closure Right The closure takes Nothing and a stream This That it's saying that the output of this needs to be a failure error Expected result Found failure error I think it's complaining here Oh Bale returns a result failure error not a failure error It is arguing with the compiler. Yeah It's basically what we cover in all these streams is arguing with the compiler It's actually just one long uh stream That is all about arguing with the compiler. Although the compiler was right here like The bale macro Actually returns. It's like return error format error But map error should not return and it should not wrap it in a result It should just return the inner thing and so that's why this has to be format error not bale And the compiler was correctly pointing that out just in a really unhelpful way in particular because like The error is on this line. It's complaining about the return type here Which is because that's where the this return type is not illegal But this return time is it would be useful if it could like point me inside and see this is where that result came from But I guess not That is true. This we knew would happen Uh because this is going to be a watched event And so the question is what kind of watched event are we expecting here? Let's see types watch So I think the event type is going to be watched event type We're expecting to first get a node created right and with Keeper state As Keeper state I guess sync connected probably Sync connected is just like everything is okay finds it connected And the server is connected. Yeah And we're expecting it to have a path and that path should be Uh slash foo So that's the event we're expecting to see Let's see if we indeed see it and down here. We're expecting to see node deleted And it's saying can't do that because It is so annoying that you cannot compare the A reference to a t with a t If t implements partial eek It's entirely unnecessary Oh, did I not derive partial eek here? Partial eek and also probably eek Partially can eek this as well partial eek and eek In theory, we could implement hash here as well, but I don't think that's particularly important. All right, let's see what happens well crashed, but got okay what trace where Proto mod 203 Well, that's interesting. I don't think the reactor should be allowed to disappear Because the threads are still running also, uh One thing that I just realized that's sort of stupid with async tests is we don't know that it's doing any of this Because it could totally be that it just like doesn't execute any of those futures and we never checked that we actually get to the end Oh, I guess we do Because otherwise this block on unwrapped would fail So I guess that's true Well, I mean it's not terribly surprising actually. So, uh, remember if you look at the top here See that the stack backtrace comes from packetizer exiting right and that I think we've seen in the past too Uh, if we scroll up to where things worked last time Uh, maybe not there, but I feel like we've Run it and had this happen before Maybe I'm wrong Because I'm fairly sure that the reactor should not exit Where is it that we have? That's right Uh, you read code for the watched event is wrong. You read code for the watched event In response, uh, why do you think it's wrong? Because I think that's because a Watcher event is a type of state in a path where these are i32s type state path i32 i32 Not in the response So in like proto somewhere Like here Watch event read from buff That should also be fine Unbounded send e No, I think this is right What what makes you think that it's wrong? Uh, sorry, what I was gonna say was there's here There's a packetizer exiting is a code that we deal with like if the spawn If we run the packetizer and it exits with an error then we print out that error And specifically what this is saying is the packetizer Future resolved with an error saying reactor gone Right and that is coming from poll read Uh specifically online. What was it? 203 Yeah, so here so when it tries to read from the underlying stream the reactor has gone away The question is how can the reactor go away because the underlying reactor shuts down on idle But it's not idle because this should still be running So that is pretty weird. I had that thought in the back of my head for too long. I've forgotten why exactly Let's see if the xid is Minus one and there was no error because that would have been labeled up here Then we parse out the watch that went from the buff And the buff is all of the bytes for that request Which seems fine And then we send the event we parsed out To the default watcher And clearly the default watcher is receiving those events and they're correct because otherwise this assert would not work correctly Right So all of the things that have to do with the watches seem to be right Uh, the question is why does the reactor crash? Or crucially, why does the reactor exit? I guess we should um just for just so that we can convince ourselves that this actually does what we want we could do Uh While we're printing stuff anyway God watcher event Let's run this with no captures Don't think it should matter but Okay, so we connect Then we call exists And then we call create Then we get the watcher event saying node created synconnected foo. So that's right We call exist again Uh, we do the delete then we got a node deleted synconnected So all this seems to be right. It's just that when the At the very end the poll read Fails with reactor. No because it's not it's still not idle Let's see. So let's uh, try to instrument this a little Uh, dropping zk For idle Just to sort of see where Yeah, so notice here that um The uh, where is it waiting for idle happens up here Right and then it takes a little while before it's shut down specifically So this is where it fails So it does fail After we call shut down on idle But why is that a failure? Uh, try to set watch to false in the last exists Here No In theory all that should mean is that we'll get another response And I don't think shut down on idle should close any ongoing connections. That seems really odd Okay, so here's what I'm going to do for now Uh packet tizer Exiting if uh E dot kind is i o Wait, isn't it i o? Do we not have i o? Oh error kind uh Other rust i Let's see. Give me i o error here somewhere So i o error So kind gives me the error kind which just should be other The question is how do I print? How do I? Easily get at the Kind other error I want to get at the error Okay, so If it's this Then let's do e dot into inner That's really unhelpful. I don't really want to format this I guess I could but it's a little silly This is a It's just error send and sync Hmm So, I mean the the really silly way to do this is if format This Is equal to or I guess even worse It's equal to Right, so this is one way of doing it where If not that So that way it's basically a way to make it be quiet in theory Except it apparently didn't work that's so weird Well, that's what I get for trying to be clever So so it is interesting because the It's deeper for my guarantee to say say not at all. You should never do what I just tried to do ever. It's a really bad idea Uh, this was more I'm trying to find a way to just quiet that particular error about the reactor going away Although the reactor just really should never go away Until we Can oh What happened I don't think so Because like Why would the reactor oh it does enter that side though? So, uh, to be clear what we're doing here is we're matching over our poll reading on our poll right We're trying to figure out whether we should exit and in this case. This is saying that poll right has exited So we've been told to exit and the right socket has been closed down At that point there really is nothing else for us to do but exit Uh And what this is saying for whatever reason even though we return not ready Even though we're turning not ready the Tokyo decides to shut down the runtime which seems like it's wrong It is totally true though that in this case uh There's no reason for us to continue Because the right socket has been closed Actually, no, that's not true. It just means that we've written out everything that's in our buffers Nothing more to write And no more requests So the question here is what do we do in this case? Uh We certainly can't send that's this is not true Uh The observation is that there it could be that there's more More data that the sukeeper servers tried to send to us I guess the question here is um, are we shutting down the outgoing connection? Because this Yeah, so so um This does not actually mean that the connection has been shut down All it means is that we've flushed our outgoing buffer And we know that there are more no more incoming requests because the incoming channel has been close to us Um And so here we could totally be nice and send to the server like look i want to close my session But we don't technically need to it's fine for us to just shut down the Channel together, but in theory we should finish reading from the client too So I guess we Implement closed session Shouldn't be too hard. Um, where's my proto here? Uh close session Really not here, but here Close session, please Yeah, so this is a very similar thing to um What we had with uh, Delete where there's no response. So I think all we want to do here It's sort of like a ping almost actually Except we might want to do it in a slightly more rigorous way Yeah, let's just do it with our requests. There's going to be an empty request here And I guess this is just going to be close Session and trying to write out a closed session request Uh close session All that really does is it writes out the off code close. I think And nothing else Empty body. I don't know what the zero here is Uh FN request zero is the xid And a watch which is going to be none. So that's all fine. So if we send a closed session, all we really do is here close session It's just one of these So here I guess we will do Oh, I see that's why we don't actually need this Uh, the reason is that because the incoming channel has been closed We're not going to send a request in this way. Anyway, we're really just going to send this off code And so this will be very similar to sending our ping In that we do this Send closed session So we're going to do this I'm gonna send a closed session So Like So Basically what we're trying to do we're gonna do all of this and then we're going to shut down the outgoing channel Uh So that The read so that the server will eventually shut down the client to us so that we'll get ready from read I guess the alternative here is that we make, uh Once we get the response we're fine. Okay, so this will be xid zero the length will be the same No xid should be zero here Uh Minus one is for watch events minus two is for pings The reason I picked zero is just because that's what was used here And the request So if you see this is in rust zookeeper at least they use xid zero for closing the session. It might not matter Um, I think we already started one if I remember correctly Uh We do not Okay, fine. So I mean, I guess we could set this to one Well, we used ah, we use zero for the connect. So we might as well use zero for the close as well Right once you get a response. I think you're allowed to reuse the Uh The id I don't think they have to be unique Uh Right, so all we're really doing is doing that and then we'll send not ready In fact, then we will Is where I want tail recursion, but I think we have to call self dot poll And the reason we have to do this is um Because we need to go through this process of of rights again This is what we wanted to avoid with pings right is that um At this point We Who actually no this is not where we should do this We should do this the moment we get exiting This stuff Self exiting here So this is this is the thing that just uh Pulse the in queue until we get no more things and once we know that Then we know that we might as well send a closed session Right And this happens before poll right. So that means we don't have to do this like self poll trick And then pull right we'll make sure that we get notified again Uh right and now in um poll right We know that here If self exiting Exiting And we have written out our entire buffer and we've flushed it then we know that there's nothing more to send And so therefore we can do self stream shutdown Which is Uh We might actually not have a way to shut down I think we do I think for async, right There's a great dot shutdown like so I guess this is really a try ready probably And so this means that eventually we're gonna I guess this might end up shutting down multiple times Which is not ideal So we're gonna have something like we're gonna need something like a Uh shutdown So so the issue here is um Imagine the shutdown fails or like shutdown is not ready. It's not able to shut down the stream Uh Then We would end up with the packetizer being pulled again and when it gets pulled again It calls pull right again and then pull right Actually, I guess because the shutdown didn't succeed It's fine for us to try to call flush again. It just means that we do a bunch extra system calls. They're sort of unnecessary So let's leave this as it is. That should be fine Uh And now in theory at least up here Uh If xid equals zero and not self first Then this is shut down And I guess I don't actually know what the server response to a shutdown close Oh, it's just an empty response, right? So There is nothing really to do Uh, so the idea is that at least in theory Uh In theory Servers should now shut down receive end Right, that's the observation that we send a closed session And then we shut down our send of the channel are the right end of the channel And now in theory the server should send the final response and then close the Close the connection as well Uh, do you really need to self pull in the case where right equals ready plus exiting? Oh, sorry. No, no, no, this was supposed to My bad This is async not ready And the reason this is not ready is because we're not actually ready until We hear the response from the server. The reason this matters is there may still be outstanding watcher events Uh, that the server Wants to notify us about and we should be nice and wait for that Expected async. All right async Expected async found item. All right, uh ready is a thing that was a thing So this is going to be Shutting down this is shutting down writer. I'm just adding some debug statements for us here so that Be able to see what's going on where's the here God responds to Close session and then I guess also down here God responds to heartbeat Uh, can you combine the last two match arms? I could I sort of want to keep this one explicit just because uh I wanted I wanted to add a comment to it and did not um We have written out actually no, you're right Still failing Um, that's because the reactor is disappearing Why on earth is the reactor disappearing? so one sort of silly thing we could try here is thread sleep Sex Just to see whether it eventually tighties up. Ooh Connection closed with four bytes left in buffer Really and pull read what God responds to close session shutting down writer Pull read Read more bytes So somehow Everything works correctly except that we get We get more bytes so it's Telling me that I get four bytes after I get the response to the closed session I want to know what those bytes are Because I think that sounds really weird Uh self dot inbox In fact, I want that out in hex really really this way Fine I get four zeros at the end The length of closed session is That long right so an empty response If I haven't if I've been paying attention then empty response Just reads nothing which in theory you should mean that it reads Uh It should mean that it reads out the xid the zxid the error code the length of the beginning Which is 16 bytes Then it gets that's when it prints the response And then it's going back around and saying that it has another four zeros No, the acknowledgement of the session closed. We already got so see how If you look here, we get this got response to closed session And that is printed Here when we get a response that has xid zero And so that's already happening um I guess actually one thing that's worth pointing out here is Uh That this is currently going to drop error if there is an error I think this is gonna I don't really want it to search I think I want Uh if let If some error Then bail uh Failed to close session response to ping just to see if that's what's going on Uh That is true It doesn't look like we get an error Right it says got response to the closed session So that is the length of 16 Wait length is 16 Is 16 Why would the length be 16? There are only three fields of Oh, the zxid is 64. Okay, so that is 16 bytes Got the response read more bytes have zero Pull right Packeters are pulled pull read read more bytes have zero Wait a second. So we go through this we need four bytes The pull read we get zero It's saying inland is not equal to zero So what is the length here? That means that we got a a length field and that length field is zero So I guess maybe just like null terminates it in its entire session. I guess I could be That's so strange server Normally sends Four At the end So yes, that worked and then the packtizer exits without an error code. I guess this might as well Print out the uh Print out the bytes that are left Let me just double check that those are actually left in the buffer Yeah, okay So let's leave that at four And now this should work just fine the question is does it still shut down the reactor if I don't sleep here? What happens? Apparently it's so weird. I have no idea why they would do that uh Yeah, it still shuts down the reactor after that exists Which is quite odd really Because this happens Got through all futures All right, just uh Yeah, I should check that it's also zeros. You're right. Uh writer got through all future because it does do that So weird, uh, you're right if uh If it is equal to zero and fine left left is this if left dot len is four and left is equal to I think I have to do this Should be the same Yep But for whatever reason The reactor is gone But as far as I understand shut down on idle It should not shut down the reactor Until all the spawned futures have exited and they clearly have not I think this is a bug in tokyo I guess let's check exactly what the docs say For shut down on idle runtime runtime Uh protocol spec for zookeeper. No, it's awful. Uh, does it may have to do with the w dot weight at the end? Um, it never gets there Well, I mean it it does but I don't think that should matter I can try it. I don't think it matters. I think it crashes before that What? Oops, that was not the command I wanted to do but luckily that didn't delete anything What? Why is that even relevant? Also notice that it's not closing correctly here now Yeah But the futures haven't all returned async ready because the Because the packetizer is not returned yet And like the very first thing we do in the packetizer is we spawn the packetizer Right, so the packetizer is still running And should continue to be pulled Because it hasn't exited yet Right, it only returns async ready if it prints packetizer done and it never prints packetizer done Oh, no, no, okay. So the reason why this works is just because It's just because the main thread exits too quickly Yeah, so you see here the if I wait at the end then I I get to see it crash If I do this the program exits too quickly for us to observe the crash It does not actually fix the issue. Uh, so the observation is that the uh, yeah, great This should not be necessary Uh, so the question is what does shutdown and idle actually do? Once it becomes idle Oh I'm being stupid Shutdown and idle returns a future that you have to wait for It does it is not a blocking call because futures aren't blocking calls Hey, okay, so we now have Exist with watchers create and delete Um Heartbeats are working. So the question now is whether we add Like get data, which currently there's no way of reading data or whether we try to add Reconnection so reconnection will get into things like zx id's because you need to Negotiate with a new server that you're connecting to Sort of what version to start reading from and get data is more sort of for completeness But it will let this return data through these channels Or we could try to start tidying up the api I think get data is probably a useful thing to add Uh Yeah, so I guess it's really up to you whether you want to see get data or reconnection Reconnection is probably going to be harder, but get data is more useful Um That's right for x and inputted code return success. Nice. Uh, isn't it what is it step three profit? Um Yeah, I think we'll probably just have time for one more because the stream is already going pretty long Um Yeah, so what do you think should we add should we add data retrieval or should we add automatic reconnection? I think those are the most feasible and useful options Uh, one other thing I'd like to point out while you have you are here is that this create as you as you observe There or there are a lot of just like functionality missing. That's not as you as you can see from when we implemented Uh Delete for example, right and create for that matter. It's actually pretty easy to add methods like that to add Additional api operations. So if you want to try your hand at this Because I'll push all the code, right Um, you should try to submit a pull request where you add if you look at the the zookeeper api for the original crate, right? Those things like set data if you look at the Java api there's a bunch more operations, right? So there's uh Get acl get children get data Get various session state multi operations setting the acl and existing nodes set data um Ooh transaction I feel like this multi thing is something we'll want to take a look at Um, but so if you want to try adding some of these then you absolutely should the other thing we really want to add is um Add all of the documentation from the java code into our crate so that the crate is actually fully self documented um What are you actually working on like outside of this or what is this? auto reconnect get data Apparently apparently the two of you, uh, highly disagree Um, so zookeeper is a you can think of it sort of as a key value store Except its goal is not so much to store data as to store metadata Uh, and it it aims to keep that data very highly consistent. Um, so if you have many Many clients that are all doing many operations at the same time It guarantees an order of operations between them and lets them detect if there are conflicts And also you can have many servers that are running as And sort of presenting a single key value store even though they're Distributed across many servers and it does the right thing even when servers fail or clients fail and those kind of things um Get children into account of future. Yeah, that's nice Uh, yeah, I mean, I suggest that you watch Part one before you watch this Uh, because this is starting pretty deep into it. You could also watch some of the other streams I've done Um, and then you read is for metadata sort of although you should think of it as high consensus um Or sorry high consistency for the data Let's see Are you a russian hacker? Uh, no But then again, wouldn't I say that if I were a russian hacker? Um, but yeah, I would watch some of the previous streams, uh, and some of those might actually give you ideas for Uh small contributions you can try to make just to get yourself familiar with it With a codebase you already know something about Uh, let's see All right, I think that Oh, that's such a good question so Get data will be fairly straightforward. So let's start auto connect Because that way one of you can hopefully maybe add get data. Um, so let's do auto connect which is going to be A little bit interesting, but we'll see how that works out. So first, uh, So, okay So one thing that's really neat is because we have this, um This sort of single packetizer business Um, we can have all reconnects handled through that So to give you some idea, there's um, where is this over here somewhere? Pa pa pa pa Okay, so this text is sort of important. Let's uh make this bigger because I hate websites that do this Uh, don't mess with my line height Uh, don't mess with my font. Don't mess with my font size Don't do any of those things Thanks Much better See how much better that is right, so the This is sort of the crucial point port that we're doing now. Um Personally learning was because Oh, yeah, I've gotten a lot of questions about actix web. Um, we might try to do something on that That host name. Oh, yeah, I'm pretty happy with my host name Um Yeah, so the idea is that when you initially create a zuki procession you get a A 64 bit number that's sort of your session id that is given. So if you look at our response The connect response It has a session id and it also has this password and the idea is that um If you ever have to re-establish a connection with a new server You send along both the session id and the the password to prove that you are that same client And so we're going to need at the very least this and this there's also this read only flag that We're going to ignore for now But the basic idea is if you're talking to a zookeeper server that has lost its connection to other zookeeper servers It's not able to do writes because it has a guarantee consistency Uh, and this it places the client in a read only mode where If you try to do a write it's not going to succeed because I can't communicate in the further And so it will just return errors and then it places you in a read only session so that you're aware of this fact And here you can use things like watchers to realize that you are now in a read only mode We're we're just going to leave that aside for now and look at this session re-establishments We'll need session id and password from the connection response um And so the idea is that when you get disconnected Then you try to connect to one of the other zookeeper servers. So remember how Uh in our source lib Currently we just take a single socket outter But in reality what you should do with zookeeper is there'll be a deployment of many servers and If you lose a connection to a server you try to reconnect a some random other server and you keep retrying until you reach a server That's that you can't actually connect to And then once you connect to some server, then you either transition to the connected state So you reconnect send its your session id and your password Um And then it responds back. Okay. Your session is good again or the server will respond saying your session has expired um And at that point Right, so the idea is that the Uh The idea is that if our session expires We're just going to try to connect to a new server and establish a new session and we'll do that I think this is saying that this should happen internally in the library and maybe that is true They say they have some like heuristics to make it more efficient. We probably won't implement those We'll just say that if we uh, if we are told that our session has expired We'll just return an error to the client saying We tried and failed um, I think is basically the idea that if if it's if at that point like You we try to re-establish the session and we couldn't re-establish it quickly enough So as session expired then good luck. You got to connect again Uh We might end up sort of making it fit more with this text later. But for now, let's leave that Uh, the other note that's important here is that we don't deal with session expression It's not like we start a timer and after a certain amount of time it is considered expired it's we try to connect and then, uh, the Clusters or the zookeeper servers, they are going to decide that we have expired And then do all deletion of all ephemeral nodes and such Uh, and then once it re-establishes contact, that's when it's told that it's uh session has expired So let's see so connected Partition then we're disconnected After a while we connect and we're told that we have expired Yep These are pings we've already dealt with that Uh, perfect. All right Ooh Session moved. Yeah, I mean we don't really care about that for now Uh, so the first thing let's just for now stick with the case of there's only one zookeeper server So we're just gonna try to do this in terms of reconnecting to the same server And then we can expand it to having a list where we just like try them round robin or something later Um, just avoid Shaking up the api too much Um, so what this really means is We're gonna have to handle Um connection errors So the way we're gonna do this is Huh, that's a very good question We sort of want to be able to continue to use the question mark operator in here So I think what we're gonna do is Um We're gonna use the new do catch notation Actually, we could have inner pole instead. Maybe that's nicer. It's a little unclear actually No, let's put it here. That's fine. So we're gonna have, uh, no not that but a Uh Pole inner Which is basically just going to be our current pole An async Item is nothing and a failure error And that is going to be this entire function like so And then what we're going to have pole do is it's going to call self pole inner And it's going to match on that If we get any kind of okay value Then we just return okay v If we get an error Then we need to be a little bit more careful Because we basically want to see if the error we get is a disconnect Then, uh We sort of want to purge state and reconnect Might not even want to purge the state. It's a little unclear because we might want to In a sense what we want to do is Any outstanding, uh, operations We need to decide what we want to do with them with we could cancel them That might be the right thing to do Yeah, I think that's the thing to do Like if we if someone does a Someone does a create and it's asynchronous, right? So we send the create and then we're notified that the server went away Uh Before we even write that request to the wire. So it's still sitting in our buffers And then we get disconnected. We reconnect to another server. Do we then send it to the create that was sent? Sort of to the previous server just never got sent or do we just prune that give an error to the user and the user has to retry I'm not entirely sure Um, this is one of the things that gets a little bit tricky in the asynchronous case and I think they say this here too that um Of course received on a connection for a session which has been re-established And I don't think this is too much of an issue Well, let's just write it up in the most straightforward way first and send Errors to the client and then over time we can make the retry happen internally potentially Um, so the question is how do we learn that the error was a disconnect? Chris goes through and gets disconnected and the server responds to the request after reconnection Um So that is one of the things that zookeeper deals with You might not hear the response Actually, that's a good question. I think what happens is you You would get the response From the server you connect to so this is one of the reasons why we have this uh zxid value So if you look at the quest in connect This is last zxid seen so every time we get a response from the server it includes a zxid right When you reconnect you tell the server what the last zxid you got from the previous server was And then uh, it will send you anything that you have missed like anything that the anything that's been recorded of something that's happened that come that You haven't you clearly have not seen and so that's how you would hear the response nonetheless The response would be resent to you If you're an original request in fact actually went through Uh, what keyboard are you using? This is a filco majesto touch ninja It's pretty nice. It has the keycaps on the bottom School Right so so that is why that is why you're guaranteed to still see those responses later on The question is how are we going to even detect that there's been a disconnect Here's one of the problems of using failure right like We don't know where that error is So let's see A disconnect would only be if it's pole read or pole write And pole read and pole write All use so there's a bail here Uh, but that is If the server actually Even in this case we sort of want to reconnect right It could be that we just want to deal with any error like this Um, but that would mean that even parse error we would start over Which doesn't seem quite right Hmm I think we'll we'll start with the simplified version Which is if there is an error we're going to reconnect because it lets us start from a clean session regardless Um Yeah, so this is one of the things that um zookeeper is cool about is the zx id basically allows you to give clients a very Strongly consistent view of the world Um And there's some rule for like When zx id's expire. So this is the idea of session expiration, right like Uh, you know that if a client session has expired Then you can remove any state that you would have sent to that client because when they connect You're just going to tell them look your state has gone and try again like establish a new session Uh, so that's the way to deal with it So what it's you can think of it as it has to keep all zx id replies within the client session expiration time Uh, I'll assume all errors are Right, so the question then is what are we going to do like what is the What is the protocol for Doing a reconnection this makes me think that we should put Handshake inside of zk, but let's um Put it here for now So certainly we'll have to remember the address. So that's going to be one thing We're going to have to like let adder is equal to something we don't quite know what yet Uh, then we're going to connect to that Well, actually I think we're going to end up making packetizer have a State that's going to be an enum of either the thing It's going to call poll inner on which is the way it progresses or if it's disconnected It's just going to poll on itself connecting So we're what we're going to do here is basically construct a future that when it resolves will become the new inner state Think of this as Uh, where is it here a state is going to be Packetizer state and we're going to have an enum Packetizer state s Which is going to contain the stream all of this stuff in fact and this stuff I think it's mostly those current state And then we're going to basically implement uh s Packetizer state s and we're going to implement something that's sort of like poll So that's basically going to be this uh most of these methods actually are going to be on packetizer state instead So let's go ahead and do that so here This is going to be connected It's going to be this a stroke packetize active packetizer It's going to be one of these And then the alternative is that we're reconnecting Those are the two states we can be in Uh, impulse same as import. No. Impel is not the same as import. Impel is a way to say that these are methods on this type so For example here If I impel enq or enq that means that if you have an enq or you can do like Dot enq to call the method enq on x So this is going to have some kind of future for reconnecting That I don't quite know what's going to be yet Uh, and then we're going to implement future for packetizer state See roughly where this is going So the idea is that you're always going to be pulling whatever is inside and whatever is inside is going to depend a little bit by I think this needs to go further down 39 So packetizer here So if you have an active packetizer, then the way to drive that forward is just to keep doing all the things If you have a Reconnecting packetizer the future you're driving forward is to reconnect And so this is why a packetizer state implementing pull for that is going to be Basically matching on self And if you are actually let's make this a box future I haven't quite decided what this will be, but I think this will be an active packetizer s and the error Will be a Failure error So if you are connected, then you have a Active packetizer and you will just call ap dot pull and if you are reconnecting Then you will do this You will do try ready ap dot pull And then here we're going to do something like Set self to be connected Right, so the idea is that if you're in the connected state Then when you're when you call pull on the packetizer state You just drive forward like you read more from the stream you write more responses you send replies those kind of things Whereas if you're in the reconnecting state, you're just going to be pulling your reconnect And then when it finishes you're going to change your own state to be connected um And this of course means that this is not going to be on packetizer state, but on active packetizer It's going to have all these methods Pull in q is the one thing it will not have because that has to happen Further up that has to happen all the way up at the packetizer level Which is actually a little awkward and then what this thing is going to do Pull inner right, so All right, we'll read so we're now implementing future for packetizer state and the packetizer has one right So if you pull a packetizer now, what's going to happen? It's going to do self dot state dot whole If I it will first do self dot Uh Pull in q Right, so we'll first do that to try to get any additional connections Which I think currently is being called up here somewhere That's going to go away. So remember they're recall that what's now left in this right? So this is what we do when we pull When we pull an active packetizer what we do is we Just pull the reader pull the writer And pull for just completion of things So we're we're sort of building a hierarchy. Let me try to start this from Higher up so it's easier to follow What's going to happen is when you pull the packetizer It will first try to see if there are more requests to get and then we'll pull its interstate The interstate is sort of a second level future That is either that we are connected and we need to drive the requests that we have outstanding forward Or we're in the reconnecting state in which case we need to continue trying to reconnect And so this is sort of the top level so implementing future for packetizer What that will do is if we're not exiting it will try to get more More requests This is going to have to do something slightly funky, which I don't know how to do yet So this is going to be yet to do And then so first we're looking for more data and then And then we will pull the The current state and if that resolves then we just give whatever result it gave and if that errors out Then that is the case where the current connection has failed as we need to retry So let's see Let's deal with the retry case a little bit later. I want to clean up some of the more stuff Further up So the packetizer is going to have to have some way To enqueue stuff That's a good question. I think this has to be There's no reason for us to pull the enqueue thread If we're currently not connected. So I think what this is going to be is If let Packetizer state connected Is self.state Only then will we pull the enqueue And in particular because then we can add things to the outbox And pull enqueue will take an ap Pull enqueue will take an active packetizer s And that way We have a way to modify the outbox, which is what we want to do We still want to use unique xids Those are not those are not per connection. They are global state The reply is we're going to cancel all the replies if we fail to disconnect If we fail to If the connection goes away All of this just writes to the outbox. So that's all fine So if we pull the packetizer, we're not exiting and we're in the connected state then we're going to Try to see if there are more requests from us Otherwise we're just going to pull the underlying state and return whatever it returns If it returns an error, then we try to reconnect This is going to do things like self. I guess We can't actually do this yet, but we will So it's going to take all the replies channels And clear them and send errors on all of them When we pull the underlying future Then if what we hit where is packetizer state down here Then when we pull that that's either going to pull the current state, which is just what used to be there So it's going to do reads and writes Or we're in the process of trying to reconnect in which case if we succeed at reconnecting Then we do a mems This is going to Probably going to come back to bite me. I think No, that would be fine Like so Then we are now connected So then we swap itself with a Accutizer state Connected with that ap and then we call self.pol. I think we need to call self.pol because otherwise When we create There's nothing that will notify the ap that it needs to be The connection that we just created needs to start doing reads and writes. Otherwise it will never be pulled again So the question is This I guess we still need to deal with we haven't actually written what the reconnect stuff will be But connected is basically the old the old packetizer, right? So that's active packetizer And all that really does I guess active packetizer is going to have to implement future If we can find it actually it's a little bit worse than that And I'll you'll see why in a second The mem swap is pretty pretty great Actually, it's not a mem swap. It's a mem replace I'm going to give you credit for catching that So let's see these things Actually have to have access to exiting And they have to have access to The where is it down here somewhere to the default watcher And neither of those are inside an active packetizer Those are all in packetizer. So we need to have a way to communicate down that into the active packetizer Yeah, so I think we're going to call this thing And that is going to be given a exiting Which is a bool Because I'm pretty sure this just reads exiting and does not write it Be very surprised if it did And it will take a Default watcher or I guess watcher which is going to be a mute Mute one of these is we'll call it default watcher um So now Where is it? self.default watcher. This is just going to use that self.exiting this is just going to be using exiting Um like so and then similarly packetizer state is not actually going to implement future Instead it's just going to be an impulse Yes Because it needs to pass along this exiting and default watcher And saying here Yeah memswap and memreplace are pretty uh pretty magical All right, this is probably going to yell at me for also some reasons probably mostly because of this reconnect business So I'm going to just comment that out and see how much it yells at me a bunch 334 It's yelling at me because This needs one of these So I'm guessing it's also yelling at me here Because this needs one of these Where else is it yelling at me? All right, we need standard mem for our memreplace trick It's telling me that default watcher is not in scope. Okay, so that's been fixed Whole right is using exiting The lat needs to be communicated there It's going to be whole read Is going to be given the default watcher Whole right is going to be given exiting That's so many errors 339 This is going to be packetizer state this 40 that's just the same error more times 72 stream right so this is going to be state uh So here's an opportunity for us to fix that but i'm just going to Not deal with this for now And say that this is going to be packetizer state Connected an active packetizer With stream timer outbox these things Reply Think those are all the things Oh, I did not like me doing that. There we go That's much fewer errors better Uh 70 who apparently it doesn't like that at all Uh, yeah, all my configs are on github if you go to github.com slash john who slash configs Right there Why is it not sent? Oh, it's because of our um box future which we haven't send first Ah, it also needs to know whether it's first Uh first can also be I think that can be here So where is our first somewhere up here? packetizer First is also part of an active packetizer for the time being Basically, whether we we've heard the first response from this particular machine that we're connected to Wait, it's escaping your thing weirdly so you're saying this is turbo fish then this Yeah, I mean this is the That is the turbo fish. This is the I guess Fat turbo fish Seems seems legit Uh Async read is not implemented for s All right, I'm missing a where s Hidden somewhere far down here this guy 346 reconnecting 410 Pull enqueue expected active packetizer Uh Pull enqueue is given a mutable reference to that. It's not a given one 436 My guesses of the escapes are because of the Uh because of the bridge that's being set up, maybe Uh because restream is doing this like synchronization of chat across all the different rooms Uh takes two parameters. That is true. It takes self dot exiting and mute self default watcher Now what mismatch types? Oh Yes, indeed. This is supposed to return. What uh just make this return a Air e because I haven't All right, so in theory, I don't think I've broken anything although I probably have Uh, I move this into pull enqueue I guess while Let that is this then do that and I guess this Should basically never happen Oh, then sure async not really So this is uh, what should we return if We try to pull enqueue, but we're not connected. I think the answer is just it's just not ready. So that's fine Oh 428 AP not found well, that's a little unclear I think it's gonna have to be this because we we basically The issue with like dealing with three connections is there's so many cases where you could not be connected anymore like here um I think we're guaranteed that this match must work because pull enqueue can't change our state and it will always return Uh, not okay not ready if uh Pull enqueue Will never return error If not connected Uh, because it will only read from the incoming channel if it is connected And it will only return error if it reads from the channel and gets an error and so therefore this can never happen Uh, yep still working on the zookeeper library. It's true async stuff So we're building an asynchronous Uh zookeeper library in rust for 10. This will not longer do this and this doesn't take this anymore 379 What do we think? Hey, it works great. So we've now just like shifted a bunch of code around to the point where we can now, uh We can in theory now have the The packetizer be in a state where it is not connected Which is all we really wanted to begin with right. So now I guess we can do uh here or something like Place ourselves in And then I will do So this is Uh prepare for reconnecting state Uh, so what do we do if we try to pull and we get an error? Well We want to Send errors to all in-flight requests Right, so that's going to be not quite clear but something like For what is replay again? Uh I could do a packetizer It's a hash map of things it's going to be for tx in Reply values I guess drain So for all of those we're going to Do something to indicate to them that their thing got canceled. In fact, maybe what we want to do is just self-state Reply clear Now this won't work because state is an enum, but let's deal with that later. Um That sends cows into the future. That's right Um clearing this will cause all the All the receivers to get a channel disconnect, which means that they will know what in fact happened that their in-flight request got canceled Um, so let's see We're gonna Dis we're gonna start a new connection Once we get that stream, we're gonna do some stuff Uh, we're not going to do this map And We're going to Here we're going to basically make the stream that's all fine Uh And then we're gonna Mem replace our favorite function self.state With in fact This won't even be needed Overwriting or dropping the old state will also cancel in-flight requests So we're going to replace that with a Uh packetizer state Reconnecting Box new Retry So retry is going to be a future right and that future is eventually going to yield our new connection So we're just going to box that up and set that that's now our state of self And we know that we're going to keep pulling it because now whenever we call self dot poll It'll call self dot state dot poll which will pull this future. So we know that it'll continue to be resolved Uh, so now the question is just what goes in here And I think the answer is basically this Now this is going to be okay Async not and then this is going to be self dot poll Uh, I think you're thinking the other way around that es6 looks a lot like rust Uh, you boxed future instead of reply. Yes. I did indeed. Good catch Let's see. So here there are a couple of things we need to make sure so this is where the zx id is going to come in so remember how Uh, we somewhere here we like decode the zx id And this we're actually going to have to keep around somewhere The problem is that's hidden all the way inside the active packet iser So I guess last zx id scene is going to be here and i64 We're going to do self dot last zx id This is going to be i64 max of self Last zx id and this I think it has to be max It's actually unclear. I think the server guarantees it sends them in order So I think it's just this Let's just to make sure say let zx id is this assert that zx id is greater than self last zx id And then that is zx id So down here now When we're reconnecting Then we're going to have to I guess up here we're going to have to do something like if let Uh, actually here package Uh connected Active packet iser there we're going to have to extract the zx id from there Is self dot state Then we're going to do all these things Um And in fact if that is not the case if it's not we're not in the connected state Then we just got to return to e because this means that there's a An error that occurred while we're trying to connect error while connecting Don't recurse For now. All right, so now we have the I guess this is last Last zx id scene So now we know that we can pass that in here. We also need to extract out the password that we got from that server um So the packet iser down here is going to have Password Which I think we're guaranteed is Always a vec. Yeah Password vec ua So we're also going to do here Uh So down here This is also going to have Revenue password And we're going to do self dot password is equal to password Gee, we're just going to do mem replace No, we're going to do mem swap. Yeah self password Right, so what we're doing here is when we get a response to the connect Then we take the password out of that connect response. We're just going to get Send back. Why is it even sent back? What does it do with the response to that? All right, I think we basically ignored that response. Yeah Uh, so that response we're going to extract the password from it and stick it in itself So that uh down here We can also do Is here ideally we would do a member place, but we can just do uh Sure, uh a ref password Come back to bite me Password is password dot split off We're gonna need nll for this which is a little sad Because um that the issue here is that The I'm now borrowing into self state Uh, which means that I won't be allowed to mem replace self state down here NLL would realize that this is okay Uh, but it's not okay Uh Similarities. Oh, I see Don't worry about it. We all uh, we all say things that we afterwards believe were silly Okay, so password gets passed along here. I guess we also need the session id Which is also here Fields for re authentication or reconnection This is going to be session id It's also an i64 And so down here When we get the connect response We also get a session id and we say self Dot session id session id Work consistent reconnect So down here, we're also going to extract out the session id from the old connection that we had We're going to reuse it for the new one In theory, we could set some timeout I think for now we're just going to leave the timeout to zero so that it's auto negotiated with the server Um And then when this eventually comes back, I guess that's the question Uh, we want to make a new packetizer, but we want to make a new active packetizer Should I think we Yeah, we want to make one of these guys Which should arguably now be a new Good wrap this in a new probably uh Not do that for now just because I'm being lazy Right, so this will now be something that resolves into an active packetizer. Uh, I guess These are all fine They all start out empty I get fine. I guess we'll add an active packetizer Or is it active packetizer down here? We'll add a FN new In terms of self Um, these fields will all be set last zxid Scene is going to be zero Session ID is going to be zero Password is going to be vek This is going to be vek new The only thing that needs to be given is a stream of s and now active packetizer so This can be replaced with new stream And same with this guy. No this guy It's gonna be new street So in theory Um, this led you into writing replies that are retry. No you did it again Uh retry That's right Yeah, I'm not gonna I'm gonna try to avoid to write that Give no guarantees Okay, so what we've done now is we've constructed a Future that will eventually resolve into an active packetizer just like what we had When you initially construct a new packetizer This worse our other new this thing, right? So this thing just really constructed a new packetizer and we're reusing the same packetizer None of that changes. We're just swapping the connection that's inside is arguably we could Instead of calling it active packetizer we could call it like Connection because that is really what it is. It's a single connection with a single zookeeper server that is currently active But active packetizer seems fine. So this is gonna do a tcp connect. It is then going to Try to connect to the active pack or try to connect and when it does Create a new active packetizer and then I guess it will also have to Somehow Send in this connection request Don't entirely know how it's going to do that because it doesn't have an enqueuer And that's what makes it a little bit odd And it sort of needs to put it at the front too Ah This I think we need to do something like Enqueue Or it's it's basically we're doing whatever this is doing In reality though, I think all of this stuff We could really stick into Uh Stick into active packetizer. It's really where it belongs So this is going to be a fn and q Take some mute self And a Request I guess item which is a request And returns Probably nothing Returns nothing. So this is going to now be all this stuff is going to be self self Uh, it will however need to know the xid which is an i32 This is going to be self It's going to return a Uh Oh, it does have to return a one-shot Sender thingy So basically the what i'm trying to achieve is that this will just call ap dot and q so Xid item And I guess it does have to give the tx Because then all of this can go away from packetizer And the tx here of course is a Is a one-shot sender Oops Does not get to return one It's going to be given one which is going to be a one-shot sender of Whatever these are This thing and these are all going to be self now. Let's go away This becomes self Right, so now This thing we can just in q directly. We're going to have give it an xid and that's going to be zero because we're just Connecting I guess it can be self dot xid Actually, we could even do this now It's going to be self dot xid And then we're going to increment it by one and we're going to do the same thing here self xid plus equals one Uh And then I guess here The main question now is whether we should Wait to make the active packetizer Uh active Until it has fully done the handshake. I don't think that matters Yeah, I mean we're basically just implementing a fancier hash table So so what we're doing here is we're connecting Setting up the packetizer Giving it the first request which is going to be the The connection request And the question is are we going to Are we going to say that the connection is ready yet? I don't think so I think we uh Oh, we sort of have to because we have to pull it that's a little weird I'm not going to lie Um All right, let me explain why this is a little weird. Um My concern is here. We're constructing a connection to the To the server. I mean the first thing we're enqueuing is the connection request, which is the right thing to do um and then The connection request is asynchronous, right? So we don't necessarily know when the response to it comes back from the server Once we switch over to Uh, well once this future resolves we're going to switch over to being in the connected state Which means that we're going to start pulling requests from the client from the user again Once we start pulling those again and start sending into the server Uh The server I guess the server is guaranteed to receive them in order, but in theory might Respond to them in different orders I don't know whether that matters So i'm worried whether we need to like wait for the connection Request to be responded to before we make it the active packetizer or mark ourselves as connected I don't think we do we do however need to give a uh, A response channel here for enqueue Right, so the enqueue that we have also takes a tx for where it should send the response to the connected uh to the To the request we're making and in this case, I think the We can just make one that we never read from Pretty sure Because if we send the tx here then we're going to drop rx Uh Don't care. I guess we could respond rx Uh e-connection Because we don't really care what's in the response because most of that is handled internally, right? So I think all we really we just like set up the receiver. We could even just drop it if we wanted to Uh, but we set it up at least so that we get to see the response And that way it's I think this is fine. Like the response will just be dropped and ignored Let's see Oh, yeah these uh The streams end up being pretty long just because there's a lot to cover Like I was sort of hoping that we might uh finish tokyo zuki for today, but there's just too much stuff It's just it's weird how There are just so many things that you end up having to deal with in the hours to start to go by uh I think we will probably Like once this compiles, I think we'll probably end it off just because It's a lot of hours and I need to like eat and stuff. Um How do I become interested in async networking? So the interesting thing is I'm not I I think there are a lot of interesting programming challenges in in asynchronous programming But it's not like I really care about async networking specifically The reason I started doing these streams were more because I wanted to show more More advanced or intermediate rust code being written so that you could get experience with writing more advanced code without necessarily having that experience yourself And so I'm sort of trying to trying to help bootstrap people's experience in that regard Uh, and I think async networking is one of those things that a lot of people are interested in and don't really know how work And so it's a good way to get a very deep dive into this And it's totally true that this is a very deep dive like most of the times if you're building things in rust You will not have to deal with these things, but the goal here was to expose like Really meaty programming stuff where you're just Basically the idea is you get to watch me fail and then see how I respond to Having to debug my code because I think that's some of the the stuff That's the most useful if you're trying to learn a new language Or we're just trying to learn programming in general is to see that other people also get stuck and then Sort of observe how they get unstuck Uh It's true as well as uh stanislav says like You sort of need things to be async and you see this in the rust world too like tokyo is just everywhere now and it's because um, and I mentioned this somewhat in my much earlier streams on Building ec2 orchestration you run into this problem where Say that you're setting up like a cluster of a hundred machines You can set up all those hundred machines with a single thread because most of what you're doing is You're sending a request to the server somewhere It's going to do a lot of work and then get back to you and you don't want the thread that's doing that work To sit or just like hang around and do nothing. It might as well do other useful work in the meantime And that's when you get into async Uh, all right, let's see where we were Proto 482 Uh, this is going to be a request I think Yep Uh 15 mech does not have field password. That is true Active packet iser does wait what uh is it? password none of this Semi password business stop with that Oh, yeah, I don't think you generally want to read async rust code if you're tipsy Just a pro tip Uh 419 Where async read and async write? 42 all right. We need an address I guess packet iser is going to have to have Uh zookeeper Heather is going to be a socket adder And I guess we're going to have to include that So And then in theory down here This is going to be 6 Yeah, I think that's right. I think synchronous is not really how the world works at this point Adder is added all clone This Sorry, there's just we're just threading things through to make the compiler happy Um 510 What is it complaining about now expected i32 Oh 492 Future is not implemented. Oh, that's right. This is a map 516 expected mute That is true Did expect me Future is not implemented for that that is fine because we changed it type mismatch Uh Yeah That's awkward Hmm So I don't know if you you saw what happened here But the the problem here is we're establishing a new tcp connection But packet iser is generic over s Whereas tcp connections are not any yes. It is some specific s Oh, it's pretty awkward. I don't know if we have a good way around that actually Um, I don't really want to make this Not generic That's a good question Uh, it's sort of like I want I want to take any s where I can establish a new one But that's not trivial either Because I mean the the one way we could do this of course is uh Also, so I could do s anything that's from tcp stream Um, the the reason it's currently s is actually because it's for a couple of reasons Uh It's because it could be that people want to do this for testing Um, like you want to test it over Like you just want to use buffers instead of having a real zookeeper running Um, or if you want to connect over unix sockets, that should also be fine Um Yeah, it's not ideal I think we're going to need our own trait here It's a little awkward Like it would basically be a trait that has an associated type, which is the address uh And It extends async read and async write and it has a connect method But I think we're going to have to Uh, I do not want to do that But I guess we have to So we're going to have a pump trait, uh zookeeper transport It's going to extend async read and async write It's going to have a type parameter adder Um It's going to have an f and connect Which takes a self adder and returns a error connect error Which implements into failure error Um, and this is going to give an a result self And self connect error and then we're going to implement zookeeper transport for Tokyo net tcp stream Which is also really weird. So the adder is going to be a socket adder in fact, it's going to be A connect is going to be And this is going to be It's actually not going to be a result. It's going to be a I don't even know if i'm allowed to do this I don't think I am Food Which has to be Into future item is self error This self connect error and this has to return. Oh, this is awful Luckily the users will basically never have to see this But it's still pretty sad So a connect feud here Oh, I can't even name the connect future I think the connect future is going to be uh Tokyo What do you get back if you do a net? Okay, so it's one of these Uh, it's going to give you a self connect feud Uh, that's true. I could take a reconnect function instead. Oh, hey Uh That might so this is sort of the the proper way to do it But you're totally right that maybe it's better to just have a reconnect future All right, just a Reconnect function. It's a little awkward though because it uh, we want it to be asynchronous So it would have to be a Function the return something that is a future although we could totally do that Uh The generics turn out pretty awkward I mean, this isn't actually that bad when you use it. It's just Having this trait in the first place is awkward Because like if you if you look at it all this is going to be is a tcp stream Connect right like that's the entire implementation for tcp stream Uh, and then down here we would say that We implement here where s is what did we call it? Uh zookeeper transport and I guess struct packetizer Is going to have the same thing Uh, because this is going to have to be s adder And now down here All we're all we really have to do is uh s Connect self adder Right, so it's actually not that bad. It's just an annoying trait to have Um, and I think there's probably yeah, this has to Where s is zookeeper transport and this has to take an s In theory, this means that we could do the connect here directly Do connect directly here now that we can connect future returns a tokyo IO error 24 Oh Info. Wow. That's when you can tell that it's been a while Uh It's high We're not going to have trait objects of this type anyway 442 This also needs to be zookeeper transport. That is true Oh, that's so many errors Uh 50 What? 58 Uh Nope, we're going to make this not be generic at all It's going to take a tokyo net tcp stream This packet is yourself even out to be generic over s and not a few methods It does because it contains an s Right, it has to contain the underlying stream s here is the stream that we're sending over And so it actually does need to be generic over s for a mod 100 Send has to be send This probably also has to be sent That is because it is into future Maybe slightly off topic and better personal preference. Does linux make it easier to do this pure program in the windows? Uh, I have not used windows for programming for Uh At least 10 years so I can't really speak to that Uh, I have never found that I wanted a different operating system for programming I have several times observed people who Use linux and a vm for programming or for particular pieces of programming Which suggests to me that I'm not really missing out But I can't speak to it directly Uh map error We did say that the error has to implement oh into failure error Uh, okay. So we have to do it this way. I'm almost there 542 with a Basically what you're doing with this transport is distracting away the stream Not quite. I'm not actually abstracting the stream away Uh, the s is doing that what I'm what I'm doing with this with the trait is I'm saying I need a stream that has certain properties So you should think of traits as like behaviors that a type supports So in this case what I'm saying is that I can be generic or packetizer can be generic over any s Where s implements async read async write sized send and has a connect method that returns Something that is a future over itself, right? So that's still packetizer being generic so I'm not I am abstracting over it in the sense that I'm allowing it to be generic, but it's not like I don't need to know what the underlying stream is. It's really the implementation doesn't care about which stream it is But it needs to know Like the types the compiler needs to know which type the stream is otherwise It couldn't put it into the struct in the first place That is also true This needs to also implement send Oh, that's not what I meant. This needs to implement send Oh, I don't know if I can even say this Like I need Into future Okay, so this is a little awkward I need so so I'm saying that connect can return anything that can be turned into a future But I need to say that the future it is turned into is send Which I can do with ware clauses, but I don't think you can have a ware clause on type So I think I actually have to do this which makes it quite awkward to write Connect Probably still fine, but Uh, and it has to be static probably Oh, right. That's our reconnecting. This has to be static So the expression is assignable where 516 It is telling me that this thing It's telling me that this thing is not Static which is true because it's borrowing self this Oh, I still have the into future. Yeah Oh, because x Xid is going to be self xid Self xid is going to be plus equal one This is going to be xid and this is okay because we know that the Will not start accepting new requests until this is finished Therefore that xid is still valid Connect feud may not live long enough Yep, because connect feud also needs to be static That's better. I like those errors better. Yeah, so this is where I told you where Uh, this is where we need nll Because I'm I'm swapping. Wait, that's not where we need 542 We're swapping self dot state But we're also borrowing out the password down here Uh, and the compiler doesn't know that that's okay Um So what I will do is this Which is all so terrible this of course now it's unreachable because we're checking it above Um, so Now we have the password down there and that's how you work around in a low Uh, oh, sorry. Yeah, you're right that we could totally store a uh box zookeeper transport, um Inside packet iser. This has always been the case, right? We could also always have Stored a box async read plus async write plus send plus static if we wanted to But there are a couple of reasons to have it be Generic one of them is to produce a faster code although that probably doesn't matter too much for zookeeper Because you don't have dynamic dispatch within of it The other reason is because it means that you can look at the type and see what kind of connection you have And this can make it easier to do things like imagine that you wanted to connect to zookeeper over tls Right, then now a part of the type will be this is a zookeeper connection over tls Which would not be the case if you boxed it. Well, that's a bunch of errors Okay, so apparently zx id's are not increasing Uh, that's also kind of interesting. Uh, zx id self lasts That's odd Oh, I see it's it's greater than or equal Might as well keep it, but Minus one G Thanks, why would zx id be minus one? All right, what does russ zookeeper do here when it sees last? It's zx id scene because that's awfully strange I don't think russ zookeeper does reconnection Yeah And if it does reconnection it does not do it correctly Because this can get minus one So unless they're Yeah, I don't think that's right I don't think this can be just be blindly accepted because what we're seeing here is I'm getting a request where the zx id is minus one I think this is a bug in upstream In the russ zookeeper crate the synchronous one I think you need to ignore things that are negative one If zx id not equal to negative one Then do this Uh, if you get negative one, then it's from the watcher What makes you say that? Because this should be a response from oh, I see I see Yeah That still makes this wrong. This line is still wrong. I guess we can assert that x id is minus Minus one Only watch events should not have zx id It's also really weird. I don't know why zx id's would not be sent for watch events Panicked a whole right why? Let's just double check to this code Not yet Just to see that that code is not actually reached No, both ping and connect disconnect should have zx id's And in fact you see that they do right like here we're getting the response to To connect I guess we should print them out here. How many response to xid? zero with opcode create session And there you see we've got zx id's 349 The exists we get 349 So this is basically how many writes they've been so a read does not increment it But you see the connect does give us zx id zero plus minus two No, zero gives us a zx id to connections does a search and failed Well, that's interesting At lib 132 153 Oh, it's because it crashed halfway through last time So I need to delete slash great I think I agree with patrick too that Now is probably a much better time to get into rust than like in the past many years I think it's also there's still things that are shaking out like the whole async await things I think tokyo still needs a bunch more work But I think there are a lot of resources. There are a lot of like the rust book is really good And so we are we are getting pretty far and at this point You can actually use rust for a lot of things like there are crates for most of the things you want to do And so I think I agree like this is a good good place to start Uh, okay, well, that's awesome Notice how the So if you look at it the initial zx id when we connect here now Let's connect Is 372 And then we do an exist and we still get 372 Then we do a create and then we get 373 because a modification happened Then we do an exist and we get 373 again because nothing changed Then we do our delete and then we get 374 because it was a change And then 374 because nothing changed 375 5 is when we leave because a a node leaving Also triggers ephemeral nodes to be deleted. And so therefore 375 hours to go up we could add a bunch of tests for things like ephemeral nodes and such I think we'll leave that for some other time We now have all the infrastructure. We haven't really tested whether reconnections work Uh, so I guess To do test this But I think that's basically I think now we have all the stuff that's in theory. This should be all that's needed for the reconnects Um There will all the clients that are waiting for us replies will indeed get errors And any that gets sent after we detect the disconnection happened those requests will still be sent as normal So I think we're in a pretty good position now So let's commit that first round Our first attempt at automatic We'll still have to do things like if you If the user gives many addresses for zookeeper Nodes and we'll have to sort of iterate through all of them and keep trying because it could be the one server You connected to went down But there are others that where the connection will work successfully We'll want some tests with watchers But I think now we have most of the sort of core protocol and the higher level api At least functional. It's not necessarily well designed yet, but it is functional Um, I think we're going to leave it there because this room is getting pretty long Um, but hopefully there'll be another one in not too long Um, so I hope you enjoyed it. I'll I'll push this um, and as always Like if you have questions or whatever then reach out to me on So I'm on twitter at twitter slash john who so you can ping me here if you Apparently someone has already If you have something you want to say there's also the patreon page Where you can follow me and subscribe If you want to hear about upcoming streams, you can also Follow on twitch or youtube I did a test for deleting non-existing things No, I so I've I've not added any test beyond the one we have There are lots of tests that could be added now because we have all the functionality and adding those is a good idea I just haven't yet. I figured I would focus more on the The functionality and the async part of things than on adding tests for everything Although that is obviously something that should be done But yeah, you can follow me on twitter follow me on youtube follow me on twitch follow me on any of the others I do stop by the rest I see chat at least every now and again. I'm the same username there I'm basically john who pretty much everywhere except twitch and youtube and mixer Which are the three streaming platforms Because on those that username was already taken in one get one of the cases taken by an old user of my own And I can't change it. So that's annoying But yeah, thanks for stopping by. I hope you found it interesting I'll post a recording of this once we're done here, and then you can go back to the places where things moved quickly So thanks for joining Bye everyone Like this