Make eth protocol handler stateful in network

[loocapro]

Closes #13955

• EthProtocolHandler generic to EthConnection
• NetworkConfig builder is able to set a generic EthProtocolHandler
• SessionManager can handle generic connections
• PendingSessionEvent can keep a generic connection (part of the ethhandler trait or boxed stream as the type)
• EthProtocolHandler with static dispatch (associated type of NetworkPrimitives or new trait that is also NetworkPrims)
• Tests

[mattsse]

before we make any significant changes to the existing session code I'd like to explore how this could look like.
we can look at the existing flow for this:

I new pending session is triggered when
a) incoming tcpstream
b) outgoing dial

this should eventually lead to a connection (Sink+Stream)

so we can design this similarly to:

reth/crates/net/network/src/protocol.rs

Lines 18 to 39 in bf20c78

	/// A trait that allows to offer additional RLPx-based application-level protocols when establishing
	/// a peer-to-peer connection.
	pub trait ProtocolHandler: fmt::Debug + Send + Sync + 'static {
	/// The type responsible for negotiating the protocol with the remote.
	type ConnectionHandler: ConnectionHandler;
	/// Invoked when a new incoming connection from the remote is requested
	///
	/// If protocols for this outgoing should be announced to the remote, return a connection
	/// handler.
	fn on_incoming(&self, socket_addr: SocketAddr) -> Option<Self::ConnectionHandler>;
	/// Invoked when a new outgoing connection to the remote is requested.
	///
	/// If protocols for this outgoing should be announced to the remote, return a connection
	/// handler.
	fn on_outgoing(
	&self,
	socket_addr: SocketAddr,
	peer_id: PeerId,
	) -> Option<Self::ConnectionHandler>;
	}

e.g. EthProtocolHandler

with the idea that the SessionManager stores an instance of this, like it does for:

reth/crates/net/network/src/session/mod.rs

Lines 110 to 111 in bf20c78

	/// Additional `RLPx` sub-protocols to be used by the session manager.
	extra_protocols: RlpxSubProtocols,

the EthProtocolHandler's input could mimic what you already have for the PendingSession

so we can do:

1. incoming tcpstream
2. create a new EthProtocolHandler::ConnectionHandler handler via on_incoming
3. await the response which is the established connection, similar to:
   

   reth/crates/net/network/src/protocol.rs

   Lines 41 to 46 in bf20c78

   	/// A trait that allows to authenticate a protocol after the `RLPx` connection was established.
   	pub trait ConnectionHandler: Send + Sync + 'static {
   	/// The connection that yields messages to send to the remote.
   	///
   	/// The connection will be closed when this stream resolves.
   	type Connection: Stream<Item = BytesMut> + Send + 'static;

[mattsse]

this likely could almost work

still need to think about type erasing the handler and whether we're missing anything but I think if we change the conn type this should work in theory

[mattsse]

for now we can keep the dyn hack, but I think we could even make this an associated type of NetworkPrimitives or a new trait that is also networkprims

[mattsse]

this is the thing that's still problematic because this is still limited to the concrete connection type that operates on EthMessage

which I think we still want because we expect those variants, but conn here is of a concrete type so we either make this part of the ethhandler trait or we use a boxed stream as the type instead