Archives
/
distant
mirror of https://github.com/chipsenkbeil/distant
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Unfinished and broken again for distant-net

Browse Source
pull/146/head
Chip Senkbeil 2 years ago
parent c68f74c4e9
commit da3b11c0db
No known key found for this signature in database
GPG Key ID: 35EF1F8EC72A4131
4 changed files with 32 additions and 49 deletions
Show Stats Download Patch File Download Diff File

30
distant-net/src/client.rs
Unescape Escape View File

@ -1,6 +1,4 @@
use crate::common::{
FramedTransport, Interest, Reconnectable, Request, Transport, UntypedResponse,
};
use crate::common::{Connection, Interest, Reconnectable, Request, Transport, UntypedResponse};
use async_trait::async_trait;
use log::*;
use serde::{de::DeserializeOwned, Serialize};
@ -44,13 +42,8 @@ where
T: Send + Sync + Serialize + 'static,
U: Send + Sync + DeserializeOwned + 'static,
{
/// Creates a client using the provided [`FramedTransport`].
///
/// ### Note
///
/// It is assumed that the provided transport has performed any necessary handshake and is
/// fully authenticated.
pub fn new<V>(mut transport: FramedTransport<V>) -> Self
/// Spawns a client using the provided [`Connection`].
fn spawn<V>(mut connection: Connection<V>) -> Self
where
V: Transport + Send + Sync + 'static,
{
@ -61,7 +54,7 @@ where
let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1);
// Ensure that our transport starts off clean (nothing in buffers or backup)
transport.clear();
connection.clear();
// Start a task that continually checks for responses and delivers them using the
// post office
@ -75,7 +68,7 @@ where
cb = reconnect_rx.recv() => {
debug!("Client got reconnect signal, so attempting to reconnect");
if let Some(cb) = cb {
let _ = match Reconnectable::reconnect(&mut transport).await {
let _ = match Reconnectable::reconnect(&mut connection).await {
Ok(()) => cb.send(Ok(())),
Err(x) => {
error!("Client reconnect failed: {x}");
@ -88,7 +81,7 @@ where
break;
}
}
result = transport.ready(Interest::READABLE | Interest::WRITABLE) => {
result = connection.ready(Interest::READABLE | Interest::WRITABLE) => {
match result {
Ok(result) => result,
Err(x) => {
@ -103,7 +96,7 @@ where
let mut write_blocked = !ready.is_writable();
if ready.is_readable() {
match transport.try_read_frame() {
match connection.try_read_frame() {
Ok(Some(frame)) => {
match UntypedResponse::from_slice(frame.as_item()) {
Ok(response) => {
@ -148,7 +141,7 @@ where
// outgoing bytes that weren't sent earlier.
if let Ok(request) = rx.try_recv() {
match request.to_vec() {
Ok(data) => match transport.try_write_frame(data) {
Ok(data) => match connection.try_write_frame(data) {
Ok(()) => (),
Err(x) if x.kind() == io::ErrorKind::WouldBlock => {
write_blocked = true
@ -166,7 +159,7 @@ where
// 1. When flush did not write any bytes, which can happen when the buffer
// is empty
// 2. When the call to write bytes blocks
match transport.try_flush() {
match connection.try_flush() {
Ok(0) => write_blocked = true,
Ok(_) => (),
Err(x) if x.kind() == io::ErrorKind::WouldBlock => write_blocked = true,
@ -199,6 +192,11 @@ where
}
impl Client<(), ()> {
/// Creates a new [`ClientBuilder`]
pub fn build() -> ClientBuilder<(), ()> {
ClientBuilder::new()
}
/// Creates a new [`TcpClientBuilder`]
pub fn tcp() -> TcpClientBuilder<()> {
TcpClientBuilder::new()

13
distant-net/src/client/builder.rs
Unescape Escape View File

@ -14,10 +14,7 @@ mod windows;
pub use windows::*;
use crate::client::Client;
use crate::common::{
authentication::{AuthHandler, Authenticate},
FramedTransport, Transport,
};
use crate::common::{authentication::AuthHandler, Connection, Transport};
use serde::{de::DeserializeOwned, Serialize};
use std::{convert, future::Future, io, time::Duration};
@ -99,12 +96,8 @@ where
let transport = self.transport;
let f = async move {
// Establish our framed transport, perform a handshake to set the codec, and do
// authentication to ensure the connection can be used
let mut transport = FramedTransport::from_client_handshake(transport).await?;
transport.authenticate(auth_handler).await?;
Ok(Client::new(transport))
let connection = Connection::client(transport, auth_handler).await?;
Ok(Client::spawn(connection))
};
match timeout {

23
distant-net/src/client/channel.rs
Unescape Escape View File

@ -134,26 +134,24 @@ where
#[cfg(test)]
mod tests {
use super::*;
use crate::client::Client;
use crate::common::{FramedTransport, InmemoryTransport};
use crate::common::{Connection, FramedTransport, InmemoryTransport};
use std::time::Duration;
use test_log::test;
type TestClient = Client<u8, u8>;
type TestChannel = Channel<u8, u8>;
/// Set up two connected transports without any handshake or authentication. This should be
/// okay since we are creating a raw client that
async fn setup(buffer: usize) -> (TestClient, FramedTransport<InmemoryTransport>) {
async fn setup(buffer: usize) -> (TestChannel, Connection<InmemoryTransport>) {
let (t1, t2) = FramedTransport::pair(buffer);
let client = TestClient::new(t1);
let channel = TestClient::new(t1);
(client, t2)
(channel, t2)
}
#[test(tokio::test)]
async fn mail_should_return_mailbox_that_receives_responses_until_transport_closes() {
let (client, mut server) = setup(100).await;
let mut channel = client.clone_channel();
let (mut channel, mut server) = setup(100).await;
let req = Request::new(0);
let res = Response::new(req.id.clone(), 1);
@ -186,8 +184,7 @@ mod tests {
#[test(tokio::test)]
async fn send_should_wait_until_response_received() {
let (client, mut server) = setup(100).await;
let mut channel = client.clone_channel();
let (mut channel, mut server) = setup(100).await;
let req = Request::new(0);
let res = Response::new(req.id.clone(), 1);
@ -202,8 +199,7 @@ mod tests {
#[test(tokio::test)]
async fn send_timeout_should_fail_if_response_not_received_in_time() {
let (client, mut server) = setup(100).await;
let mut channel = client.clone_channel();
let (mut channel, mut server) = setup(100).await;
let req = Request::new(0);
match channel.send_timeout(req, Duration::from_millis(30)).await {
@ -217,8 +213,7 @@ mod tests {
#[test(tokio::test)]
async fn fire_should_send_request_and_not_wait_for_response() {
let (client, mut server) = setup(100).await;
let mut channel = client.clone_channel();
let (mut channel, mut server) = setup(100).await;
let req = Request::new(0);
match channel.fire(req).await {

15
distant-net/src/server.rs
Unescape Escape View File

@ -220,10 +220,8 @@ where
mod tests {
use super::*;
use crate::common::{
authentication::{
Authenticate, AuthenticationMethod, DummyAuthHandler, NoneAuthenticationMethod,
},
FramedTransport, InmemoryTransport, MpscListener, Request, Response,
authentication::{AuthenticationMethod, DummyAuthHandler, NoneAuthenticationMethod},
Connection, InmemoryTransport, MpscListener, Request, Response,
};
use async_trait::async_trait;
use std::time::Duration;
@ -286,18 +284,17 @@ mod tests {
.expect("Failed to start server");
// Perform handshake and authentication with the server before beginning to send data
let mut transport = FramedTransport::from_client_handshake(transport)
let mut connection = Connection::client(transport, DummyAuthHandler)
.await
.unwrap();
transport.authenticate(DummyAuthHandler).await.unwrap();
.expect("Failed to connect to server");
transport
connection
.write_frame(Request::new(123).to_vec().unwrap())
.await
.expect("Failed to send request");
// Wait for a response
let frame = transport.read_frame().await.unwrap().unwrap();
let frame = connection.read_frame().await.unwrap().unwrap();
let response: Response<String> = Response::from_slice(frame.as_item()).unwrap();
assert_eq!(response.payload, "hello");
}

Write Preview
Loading…
Cancel
Save