tokio_util::either::Either for handshake Io

[Antiarchitect]

Trying to use one handshake callback for both TlsStream and TcpStream via tokio_util::either::Either:

async fn handshake(
    handshake: v3::Handshake<Either<TlsStream<TcpStream>, TcpStream>>,
) -> Result<v3::HandshakeAck<Either<TlsStream<TcpStream>, TcpStream>, ClientSession>, ServerError> {
    let session = ClientSession {
        id: Uuid::new_v4(),
        sink: handshake.sink(),
    };
    Ok(handshake.ack(session, false))
}

    let _server = ntex::server::Server::build()
        .bind("mqtts", &*MQTT_SERVER_SECURE_ADDRESS, move || {
            pipeline_factory(tls_acceptor.clone())
                .map_err(|_err| MqttError::Service(ServerError {}))
                .map(|io| Either::Left(io))
                .and_then(
                    MqttServer::new().v3(v3::MqttServer::new(handshake)
                        .control(fn_factory_with_config(
                            |session: v3::Session<ClientSession>| async move {
                                Ok::<_, ServerError>(fn_service(move |req| {
                                    control_v3(session.clone(), req)
                                }))
                            },
                        ))
                        .publish(fn_factory_with_config(
                            |session: v3::Session<ClientSession>| async move {
                                Ok::<_, ServerError>(fn_service(move |req| {
                                    publish_v3(session.clone(), req)
                                }))
                            },
                        ))),
                )
        })
        .expect("Cannot bind to secure address!")
        .bind("mqtt", &*MQTT_SERVER_INSECURE_ADDRESS, move || {
            MqttServer::new().v3(v3::MqttServer::new(handshake)
                .control(fn_factory_with_config(
                    |session: v3::Session<ClientSession>| async move {
                        Ok::<_, ServerError>(fn_service(move |req| {
                            control_v3(session.clone(), req)
                        }))
                    },
                ))
                .publish(fn_factory_with_config(
                    |session: v3::Session<ClientSession>| async move {
                        Ok::<_, ServerError>(fn_service(move |req| {
                            publish_v3(session.clone(), req)
                        }))
                    },
                )))
        })
        .expect("Cannot bind to insecure address!")
        .workers(*MQTT_SERVER_WORKERS)
        .run()
        .await;

Successfully mapped io into Either::Left in case of Tls, but cannot do the same in case of Tcp server. Do not know how.

[Antiarchitect]

Found the solution:

...
        .bind("mqtt", &*MQTT_SERVER_INSECURE_ADDRESS, || {
            pipeline_factory(move |io| async move { Ok(io) })
                .map(Either::Right)
                .and_then(
                    MqttServer::new().v3(v3::MqttServer::new(handshake)
                        .control(fn_factory_with_config(
                            |session: v3::Session<ClientSession>| async move {
                                Ok::<_, ServerError>(fn_service(move |req| {
                                    control_v3(session.clone(), req)
                                }))
                            },
                        ))
                        .publish(fn_factory_with_config(
                            |session: v3::Session<ClientSession>| async move {
                                Ok::<_, ServerError>(fn_service(move |req| {
                                    publish_v3(session.clone(), req)
                                }))
                            },
                        ))),
                )
        })
        .expect("Cannot bind to insecure address!")
...

[fafhrd91]

can't you just use T instead of specific type?

async fn handshake(handshake: v3::Handshake<T>) -> Result<v3::HandshakeAck<T, ClientSession>, ServerError> {
  ...
}

[fafhrd91]

I see, ok

[fafhrd91]

btw maybe you should use different handshake service in case of ssl?

[Antiarchitect]

Tried, looks ugly because all the stuff (session creation and storage) stays the same but the authentication mechanism.

[fafhrd91]

you know that you can combine services? like future combinators

pipeline_factory(auth_service).and_then(session_creation_service)

session_creation_service could be the same for both servers

[Antiarchitect]

Thanks for the tip!