#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
use crate::error::Error;
use bytes::Bytes;
use futures::future::BoxFuture;
use futures::prelude::*;
use libp2p_core::upgrade::{InboundConnectionUpgrade, OutboundConnectionUpgrade};
use libp2p_core::UpgradeInfo;
use libp2p_identity as identity;
use libp2p_identity::PeerId;
use libp2p_identity::PublicKey;
use std::{
io, iter,
pin::Pin,
task::{Context, Poll},
};
mod error;
mod handshake;
mod proto {
#![allow(unreachable_pub)]
include!("generated/mod.rs");
pub(crate) use self::structs::Exchange;
}
#[derive(Clone)]
pub struct Config {
local_public_key: identity::PublicKey,
}
impl Config {
pub fn new(identity: &identity::Keypair) -> Self {
Self {
local_public_key: identity.public(),
}
}
}
impl UpgradeInfo for Config {
type Info = &'static str;
type InfoIter = iter::Once<Self::Info>;
fn protocol_info(&self) -> Self::InfoIter {
iter::once("/plaintext/2.0.0")
}
}
impl<C> InboundConnectionUpgrade<C> for Config
where
C: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
type Output = (PeerId, Output<C>);
type Error = Error;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_inbound(self, socket: C, _: Self::Info) -> Self::Future {
Box::pin(self.handshake(socket))
}
}
impl<C> OutboundConnectionUpgrade<C> for Config
where
C: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
type Output = (PeerId, Output<C>);
type Error = Error;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_outbound(self, socket: C, _: Self::Info) -> Self::Future {
Box::pin(self.handshake(socket))
}
}
impl Config {
async fn handshake<T>(self, socket: T) -> Result<(PeerId, Output<T>), Error>
where
T: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
tracing::debug!("Starting plaintext handshake.");
let (socket, remote_key, read_buffer) = handshake::handshake(socket, self).await?;
tracing::debug!("Finished plaintext handshake.");
Ok((
remote_key.to_peer_id(),
Output {
socket,
remote_key,
read_buffer,
},
))
}
}
pub struct Output<S>
where
S: AsyncRead + AsyncWrite + Unpin,
{
pub socket: S,
pub remote_key: PublicKey,
read_buffer: Bytes,
}
impl<S: AsyncRead + AsyncWrite + Unpin> AsyncRead for Output<S> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<Result<usize, io::Error>> {
if !self.read_buffer.is_empty() {
let n = std::cmp::min(buf.len(), self.read_buffer.len());
let b = self.read_buffer.split_to(n);
buf[..n].copy_from_slice(&b[..]);
return Poll::Ready(Ok(n));
}
AsyncRead::poll_read(Pin::new(&mut self.socket), cx, buf)
}
}
impl<S: AsyncRead + AsyncWrite + Unpin> AsyncWrite for Output<S> {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, io::Error>> {
AsyncWrite::poll_write(Pin::new(&mut self.socket), cx, buf)
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
AsyncWrite::poll_flush(Pin::new(&mut self.socket), cx)
}
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
AsyncWrite::poll_close(Pin::new(&mut self.socket), cx)
}
}