client: add Subscription::close_reason

[niklasad1]

This PR changes adds a new API to the subscription know how the subscription was closed.

It still drops the subscription if it's lagging but the subscription should easily-extendable to
implement specific handling for lagging, see the example below

Example of custom subscription logic

fn drop_oldest_when_lagging<T: Clone + DeserializeOwned + Send + Sync + 'static>(
	mut sub: Subscription<T>,
	buffer_size: usize,
) -> impl Stream<Item = T> {
	let (mut tx, rx) = async_broadcast::broadcast(buffer_size);

	tx.set_overflow(true);

	tokio::spawn(async move {
		// poll sub, sending msgs to our chan which throws stuff away
		while let Some(n) = sub.next().await {
			let msg = match n {
				Ok(msg) => msg,
				Err(e) => {
					tracing::error!("Failed to recv subscription message: {e}");
					continue;
				}
			};

			tx.try_broadcast(msg).expect("Infallible in overflowing mode; qed");
		}
	});
	rx
}

Resolves #1291

[jsdw]

Ooh I didn't realise that this channel handled lagging messages, TIL :)

[jsdw]

Cool! Looks good to me :)

[lexnv]

LGTM! Nice one 👍

With this PR we should have a bit more info about what happens when the subscription stream ends, which makes our life easier when debugging connectivity issues 🙏