Don't require an Encoder type while constructing FramedWrite so we ca…

…n be generic over the lifetime

CHANGES.md

@@ -1,7 +1,8 @@
 # CHANGES
 
 ## Unreleased - 2021-xx-xx
-
+* Changed the signature of `actix::io::FramedWrite` to allow `FramedWrite::write()`'s `item`
+  lifetime to be the lifetime of the function rather than the struct. [#462]
 
 ## 0.11.0-beta.1 - 2021-01-01
 ### Added

Cargo.toml

@@ -33,7 +33,7 @@ derive = ["actix_derive"]
 mailbox_assert = []
 
 [dependencies]
-actix-rt = "2.0.0-beta.1"
+actix-rt = "=2.0.0-beta.2"
 actix_derive = { version = "0.5", optional = true }
 actix-macros = { version = "0.1.3", features = ["actix-reexport"] }
 

examples/chat/src/client.rs

@@ -5,7 +5,7 @@ use std::{io, net, thread};
 use actix::prelude::*;
 use tokio::io::WriteHalf;
 use tokio::net::TcpStream;
-use tokio_util::codec::FramedRead;
+use tokio_util::codec::{Encoder, FramedRead};
 
 mod codec;
 
@@ -45,9 +45,9 @@ async fn main() {
 
 struct ChatClient {
     framed: actix::io::FramedWrite<
-        codec::ChatRequest,
         WriteHalf<TcpStream>,
         codec::ClientChatCodec,
+        <codec::ClientChatCodec as Encoder<codec::ChatRequest>>::Error,
     >,
 }
 

examples/chat/src/session.rs

@@ -8,6 +8,7 @@ use actix::prelude::*;
 
 use tokio::io::WriteHalf;
 use tokio::net::TcpStream;
+use tokio_util::codec::Encoder;
 
 use crate::codec::{ChatCodec, ChatRequest, ChatResponse};
 use crate::server::{self, ChatServer};
@@ -29,7 +30,11 @@ pub struct ChatSession {
     /// joined room
     room: String,
     /// Framed wrapper
-    framed: actix::io::FramedWrite<ChatResponse, WriteHalf<TcpStream>, ChatCodec>,
+    framed: actix::io::FramedWrite<
+        WriteHalf<TcpStream>,
+        ChatCodec,
+        <ChatCodec as Encoder<ChatResponse>>::Error,
+    >,
 }
 
 impl Actor for ChatSession {
@@ -125,7 +130,11 @@ impl Handler<Message> for ChatSession {
 impl ChatSession {
     pub fn new(
         addr: Addr<ChatServer>,
-        framed: actix::io::FramedWrite<ChatResponse, WriteHalf<TcpStream>, ChatCodec>,
+        framed: actix::io::FramedWrite<
+            WriteHalf<TcpStream>,
+            ChatCodec,
+            <ChatCodec as Encoder<ChatResponse>>::Error,
+        >,
     ) -> ChatSession {
         ChatSession {
             addr,

src/io.rs

@@ -306,17 +306,17 @@ where
 
 /// A wrapper for the `AsyncWrite` and `Encoder` types. The AsyncWrite will be flushed when this
 /// struct is dropped.
-pub struct FramedWrite<I, T: AsyncWrite + Unpin, U: Encoder<I>> {
+pub struct FramedWrite<T: AsyncWrite + Unpin, U, E: From<io::Error> = std::io::Error> {
     enc: U,
-    inner: UnsafeWriter<T, U::Error>,
+    inner: UnsafeWriter<T, E>,
 }
 
-impl<I, T: AsyncWrite + Unpin, U: Encoder<I>> FramedWrite<I, T, U> {
+impl<T: AsyncWrite + Unpin, U, E: From<io::Error>> FramedWrite<T, U, E> {
     pub fn new<A, C>(io: T, enc: U, ctx: &mut C) -> Self
     where
-        A: Actor<Context = C> + WriteHandler<U::Error>,
+        A: Actor<Context = C> + WriteHandler<E>,
         C: AsyncContext<A>,
-        U::Error: 'static,
+        E: 'static,
         T: Unpin + 'static,
     {
         let inner = UnsafeWriter(
@@ -343,9 +343,9 @@ impl<I, T: AsyncWrite + Unpin, U: Encoder<I>> FramedWrite<I, T, U> {
 
     pub fn from_buffer<A, C>(io: T, enc: U, buffer: BytesMut, ctx: &mut C) -> Self
     where
-        A: Actor<Context = C> + WriteHandler<U::Error>,
+        A: Actor<Context = C> + WriteHandler<E>,
         C: AsyncContext<A>,
-        U::Error: 'static,
+        E: 'static,
         T: Unpin + 'static,
     {
         let inner = UnsafeWriter(
@@ -390,7 +390,10 @@ impl<I, T: AsyncWrite + Unpin, U: Encoder<I>> FramedWrite<I, T, U> {
     }
 
     /// Writes an item to the sink.
-    pub fn write(&mut self, item: I) {
+    pub fn write<I>(&mut self, item: I)
+    where
+        U: Encoder<I, Error = E>,
+    {
         let mut inner = self.inner.0.borrow_mut();
         let _ = self.enc.encode(item, &mut inner.buffer).map_err(|e| {
             inner.error = Some(e);
@@ -404,9 +407,19 @@ impl<I, T: AsyncWrite + Unpin, U: Encoder<I>> FramedWrite<I, T, U> {
     pub fn handle(&self) -> SpawnHandle {
         self.inner.0.borrow().handle
     }
+
+    /// Returns a reference to the underlying encoder.
+    pub fn encoder(&self) -> &U {
+        &self.enc
+    }
+
+    /// Returns a mutable reference to the underlying encoder.
+    pub fn encoder_mut(&mut self) -> &mut U {
+        &mut self.enc
+    }
 }
 
-impl<I, T: AsyncWrite + Unpin, U: Encoder<I>> Drop for FramedWrite<I, T, U> {
+impl<T: AsyncWrite + Unpin, U, E: From<io::Error>> Drop for FramedWrite<T, U, E> {
     fn drop(&mut self) {
         // Attempts to write any remaining bytes to the stream and flush it
         let mut async_writer = self.inner.1.borrow_mut();