Here's a question: I feel like I started wading into rxjs with a very hard problem, websocket connections. Websockets are shared resources, and so need to be modeled using a subject somewhere along the line. They're also bi-directional, and so I'm not sure how to pass them around in an idiomatic way. I'd like to define a function like`implementAuthPolicy(tx, rx)` that can intercept and buffer both sent and received messages. Maybe that's less of a question than me just thinking out loud.
Highly recommend rxjs, it's been an incredible journey for me and I had to learn some things the hard way.
After a heavy year on rxjs, I manage to build very complex pipelines with have very little state (variables) and close to no imperative code, I kinda got to a point of no way back.
Your approach of takeUntil is totally correct, which is the rxjs way of doing things, a powerful advice to rxjs is to avoid Subjects at all costs, they deviate the concepts of reactive programming and forces you in the imperative approach, people using rxjs with only Subjects aren't really using rxjs at all, just a fancy event emitter (the meme), there's a reason why xstream (a alternative library to rxjs) called their `next` function `shamefullySendNext` (yes, there's use cases for Subjects)
I have no issues with typescript in rxjs, it works pretty well and point-free operators would work just fine, I have issues naming things properly.
You have to be careful as well where you subscribe to the observables, it should be the final stage and usually close to the UI, and unsubscribe when the ui is "unmounted".
I was writing this document for nostr:nprofile1qydhwumn8ghj7umpw3jkcmrfw3jju6r6wfjrzdpe9e3k7mgpz9mhxue69uhkummnw3e82efwvdhk6qg5waehxw309aex2mrp0yhxgctdw4eju6t0qyxhwumn8ghj7mn0wvhxcmmvqyg8wumn8ghj7mn0wd68ytnhd9hx2qpqye5ptcxfyyxl5vjvdjar2ua3f0hynkjzpx552mu5snj3qmx5pzjs7haql4 the other day but never had a change to finish, which is my client approach for nostr subscription in rxjs https://github.com/cesardeazevedo/nostr-observable, the idea was to have a 100% stateless library for nostr.
This file is how I handle subscription and everything else is built on top of it https://github.com/cesardeazevedo/nosotros/blob/main/src/core/operators/subscribe.ts
If you have any questions don't hesitate to ask, I would love to help anything rxjs related.
Discussion
Working with websocketSubjects has been very easy and flexible, I can even push messages before the websocket was connected and the subject will buffer after connected.
webSocketSubjects just like the other subjects are multicasted.
this will only open 1 connection no matter how many subscriptions you have
```ts
const relay = webSocketSubject('wss://relay1.com')
relay.subscribe()
relay.subscribe()
```
now creating multiple webSocketSubjects it will open multiple connections.
webSocketSubjects multiplex will make a unicast subscription, this will make multiple nostr subscriptions in the same connection.
```ts
const relay = webSocketSubject('wss://relay1.com')
const sub = relay.multiplex(() => ["REQ",
sub.subscribe()
sub.subscribe()
```
You can pass the webSocketSubject as you wish as it's also a observable, I store relays as webSocketSubjects in a Map
Current my core/pool is doing the relay subscription which is incorrect, the consumer should be responsible to initiate the relay connection because he could put any policy mechanism for a particular relay, something like a rxjs retry to reconnect or some auth policy, currently my pool exposes a open callback and we could do something like this (pseudo code)
I am still working on auth and trying to think about some flexible way to do it, I'm a little bit hesitant to replay old messages after a a successful auth as I need to store these messages somewhere, but here's a small authenticator (which I think I should rename to createAuthPolicy or something) what takes a pool, a signer a options like whitelist which is an observable as the user can whitelist a observable after the auth was received.
Still half backed as I am still testing locally
Yeah, the tricky part seems to be intercepting send messages so that you can buffer them and re-send after authentication (or remove reqs that have been closed before they were sent while you were waiting for auth). Here's my (half-baked) policy for that:
```
export const socketPolicyDeferOnAuth = (socket: Socket) => {
const send$ = socket.send$
const buffer: ClientMessage[] = []
const authState = new AuthState(socket)
const okStatuses = [AuthStatus.None, AuthStatus.Ok]
// Defer sending certain messages when we're not authenticated
socket.send$ = send$.pipe(
filter(message => {
// Always allow sending auth
if (isClientAuth(message)) return true
// Always allow sending join requests
if (isClientEvent(message) && message[1].kind === AUTH_JOIN) return true
// If we're not ok, remove the message and save it for later
if (!okStatuses.includes(authState.status)) {
buffer.push(message)
return false
}
return true
}),
)
// Send buffered messages when we get successful auth
return authState.subscribe((status: AuthStatus) => {
if (okStatuses.includes(status)) {
const reqs = new Set(buffer.filter(isClientReq).map(nth(1)))
const closed = new Set(buffer.filter(isClientClose).map(nth(1)))
for (const message of buffer.splice(0)) {
// Skip requests that were closed before they were sent
if (isClientReq(message) && closed.has(message[1])) continue
// Skip closes for requests that were never sent
if (isClientClose(message) && reqs.has(message[1])) continue
socket.send(message)
}
}
})
}
```
Thanks for the code, just doing some quick code review here.
I think the first issue I spotted is the assigning a stream send$ to itself `socket.send$ = send$.pipe` try to think some other way to do that, it's very easy to compose multiple observables that derives from the same source in a functional style.
I would do something like on socket.ts
```ts
this.socket = webSocketSubject('...')
this.events$ = socket.pipe(filter(msg => msg[0] === 'event'))
this.auths$ = socket.pipe(filter(msg => msg[0] === 'AUTH'))
this.authsJoin$ = auths$.pipe(filter(msg => msg[1].kind === 28934))
this.oks$ = socket.pipe(filter(msg => msg[0] === 'OK'))
```
I would also remove the `authState.subscribe` for something like `return authState.pipe(tap(() => {}))`, but AuthState is a subject with a bunch of subscriptions inside there... hmmm, that's a bit bad, by looking at the commit history you basically kinda ported a event emitter system to rxjs right? you gonna have to embrace some heavy changes to see the gains of rxjs.
Try to think this way, let's say you have a single nostr subscription to a single relay, this subscription has all sorts of complex policies attached to the main stream, the main stream is also connected with other subscription stream like getting each user for each note in the main stream with also their own set of policies, then you decided to call unsubscribe() from the main stream, then the entire pipeline is unsubscribed and garbage collected, nothing left, with your current approach I am not seeing this happening, you might ended up with some parts of stream still listening for changes, unless you complete/unsubscribe the streams manually in multiple places which is a lot of work, which is something you have to do with event emitters.
I also did these exact mistakes a year ago, so I really don't judge you 😅
this angular tutorial describes really well the issue.
Yeah, I think I understand all that in the abstract, I'm just finding it quite hard to actually execute, partly because of the duplex nature of websockets, which is what I've been focusing on far, and partly due to the forking nature of streams, which makes me unsure what subscriptions need to be managed and which don't.
RxJs is a mental journey, so embrace it haha, I know you do a bunch of functional programming throughout all coracle code, rxjs will just fit if you embrace it's declarative approach, like your `await tryCatch` function is kinda rxjs `catchError`.
If you have any other question or want to schedule a call just let me know
Thanks, I might just do that. For now, I'm probably going to go with emitter version since I really need to get this refactor done. But let me know when you publish the rxjs relay stuff, I'd love to learn/contribute there.
This gif is also really nice.
https://video.nostr.build/e2536a24789e20584e6d49cce9c078a99a75ec0a58e3b0c638931584a8117bbb.mp4
nostr:npub1cesrkrcuelkxyhvupzm48e8hwn4005w0ya5jyvf9kh75mfegqx0q4kt37c a few weeks back I threw this together as a exercise, It has some bugs with reconnecting and DOS the relay but I liked how the NIP-42 auth turned out and that I was able to write it in <200 lines
https://github.com/hzrd149/applesauce/blob/master/packages/relay/src/relay.ts
I'm not really part of this conversation and I'm distracted by other projects but I still really want to help build a pure rxjs relay connection library 😁
I do too. nostr:nprofile1qyg8wumn8ghj7mn0wd68ytnddakj7qgwwaehxw309ahx7uewd3hkctcpzemhxue69uhkummnw3ex2mrfw3jhxtn0wfnj7qg4waehxw309aex2mrp0yhxgctdw4eju6t09uq3vamnwvaz7tmjv4kxz7fwd4hhxarj9ec82c30qqsvvcpmpuwvlmrztkwq3d6nunmhf6hh688jw6fzxyjmtl2d5u5qr8sq7rxj5 is there a repository with wip code anywhere?
wow, this is looking really good, just found weird req returns "EOSE" | NostrEvent, If I want to listen for "EOSE" I would listen on subscribe({complete: () => {}}) or `finalize` operator, but might going to steal you waitForAuth, I am think I am gonna need it 🤣
nostr:npub1cesrkrcuelkxyhvupzm48e8hwn4005w0ya5jyvf9kh75mfegqx0q4kt37c nostr:npub1jlrs53pkdfjnts29kveljul2sm0actt6n8dxrrzqcersttvcuv3qdjynqn I did a little more work on my implementation and id like to get your feedback. Its still missing re-connection logic but personally I really excited with how it turned out
https://github.com/hzrd149/applesauce/tree/master/packages/relay
That's a slick interface, I like it a lot. The main reason I wouldn't use this implementation is that a lot of the policy is baked in, which might need to be configurable by the application. I have a bunch of policy stuff here: https://github.com/coracle-social/welshman/blob/net2/packages/net2/src/policy.ts which is mostly reasonable defaults, but if users wanted to configure auth policy per relay I don't see how that could be done in applesauce. Also, why no`subscribe` ?
My idea is that it could be built on top if the relay communication is flexible enough. most nostr libraries build the auth, limitations, and connection policies at the "new WebSocket()" level, but that makes it difficult to add additional types of behavior that a different client might want later
For example you could implement a relay blacklist at the websocket level so that it would throw an error (or make it look offline) if the client connected to a blacklisted relay. or you could extend that relay pool class to ensure it never even tries to connect to blacklisted relays. all of this is just a theory though, I still have to put it into practice to see if it works
