4

SMTP 服务器应在建立连接时显示欢迎消息（220 服务就绪），这是客户端开始发送命令的信号。这似乎与 tokio-proto 的请求-响应范式相冲突。

我可以想象协议可以完全颠倒，例如服务器发送请求和客户端响应（不推荐使用的 TURN），但目前我只关心连接时的欢迎消息，也就是横幅。之后，客户端请求 => 服务器响应将被支持。

我一直在想把它挂在哪里，但是bind_server,bind_transport对我来说是超级神秘的。我需要实施运输吗？

我在decode编解码器的方法中有这个。问题是decode除非有可用于解码哪种有意义的数据，否则不会调用该方法。我希望有一些连接初始化方法可以挂钩，但我什么也没找到。

fn decode(&mut self, buf: &mut BytesMut) -> Result {

    if !self.initialized {
        println!(
            "new connection from {:?} to {:?}",
            self.peer_addr,
            self.local_addr
        );

        self.requests.push(SmtpCommand::Connect {
            local_addr: self.local_addr,
            peer_addr: self.peer_addr,
        });

        self.initialized = true;
    }
    //... snip
    match self.requests.is_empty() {
        true => Ok(None),
        false => Ok(Some(self.requests.remove(0))),
    }
}

我正在进行的研究项目在 GitHub 上，我还打开了一个关于 tokio-proto 的问题。

4

1 回答 1

0

实现我自己的有状态传输装饰器 ( SmtpConnectTransport) 就成功了。它将在初始化时注入给定的帧。我想它可以通过将 initframe 类型作为参数制成一个通用解决方案。除了解析和序列化之外，最后的编解码器不必做任何不寻常的事情。

连接后立即出现帧，该服务可以生成所需的欢迎消息或横幅。为了服务的利益，我已将本地和远程套接字地址包含在其中SmtpCommand::Connect，因为它将用于垃圾邮件检测。

我的预感是正确的，但解决它确实感觉像是生锈的金属研磨：D 我现在很高兴这个samotop 是如何组合在一起的。这是一些代码：

use std::io;
use std::str;
use bytes::Bytes;
use model::response::SmtpReply;
use model::request::SmtpCommand;
use protocol::codec::SmtpCodec;
use tokio_proto::streaming::pipeline::{Frame, Transport, ServerProto};
use tokio_io::codec::Framed;
use futures::{Stream, Sink, StartSend, Poll, Async};
use protocol::parser::SmtpParser;
use protocol::writer::SmtpSerializer;

type Error = io::Error;
type CmdFrame = Frame<SmtpCommand, Bytes, Error>;
type RplFrame = Frame<SmtpReply, (), Error>;

pub struct SmtpProto;

impl<TIO: NetSocket + 'static> ServerProto<TIO> for SmtpProto {
    type Error = Error;
    type Request = SmtpCommand;
    type RequestBody = Bytes;
    type Response = SmtpReply;
    type ResponseBody = ();
    type Transport = SmtpConnectTransport<Framed<TIO, SmtpCodec<'static>>>;
    type BindTransport = io::Result<Self::Transport>;

    fn bind_transport(&self, io: TIO) -> Self::BindTransport {
        // save local and remote socket address so we can use it as the first frame
        let initframe = Frame::Message {
            body: false,
            message: SmtpCommand::Connect {
                local_addr: io.local_addr().ok(),
                peer_addr: io.peer_addr().ok(),
            },
        };
        let codec = SmtpCodec::new(
            SmtpParser::session_parser(),
            SmtpSerializer::answer_serializer(),
        );
        let upstream = io.framed(codec);
        let transport = SmtpConnectTransport::new(upstream, initframe);
        Ok(transport)
    }
}

pub struct SmtpConnectTransport<TT> {
    initframe: Option<CmdFrame>,
    upstream: TT,
}

impl<TT> SmtpConnectTransport<TT> {
    pub fn new(upstream: TT, initframe: CmdFrame) -> Self {
        Self {
            upstream,
            initframe: Some(initframe),
        }
    }
}

impl<TT> Stream for SmtpConnectTransport<TT>
where
    TT: 'static + Stream<Error = Error, Item = CmdFrame>,
{
    type Error = Error;
    type Item = CmdFrame;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        match self.initframe.take() {
            Some(frame) => {
                println!("transport initializing");
                Ok(Async::Ready(Some(frame)))
            }
            None => self.upstream.poll(),
        }
    }
}

impl<TT> Sink for SmtpConnectTransport<TT>
where
    TT: 'static + Sink<SinkError = Error, SinkItem = RplFrame>,
{
    type SinkError = Error;
    type SinkItem = RplFrame;

    fn start_send(&mut self, request: Self::SinkItem) -> StartSend<Self::SinkItem, io::Error> {
        self.upstream.start_send(request)
    }

    fn poll_complete(&mut self) -> Poll<(), io::Error> {
        self.upstream.poll_complete()
    }

    fn close(&mut self) -> Poll<(), io::Error> {
        self.upstream.close()
    }
}

impl<TT> Transport for SmtpConnectTransport<TT>
where
    TT: 'static,
    TT: Stream<Error = Error, Item = CmdFrame>,
    TT: Sink<SinkError = Error, SinkItem = RplFrame>,
{
}


pub trait NetSocket: AsyncRead + AsyncWrite {
    fn peer_addr(&self) -> Result<SocketAddr>;
    fn local_addr(&self) -> Result<SocketAddr>;
}

impl NetSocket for TcpStream {
    fn peer_addr(&self) -> Result<SocketAddr> {
        self.peer_addr()
    }
    fn local_addr(&self) -> Result<SocketAddr> {
        self.local_addr()
    }
}
于 2017-10-22T18:54:05.390 回答