diff --git a/src/main.rs b/src/main.rs
index c152f35..6027f07 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -56,7 +56,7 @@ async fn main() -> Result<()> {
 }
 
 async fn handle_socket(websocket: WebSocket) -> Result<()> {
-    let (sink, mut source) = websocket.split();
+    let (sink, source) = websocket.split();
 
     let mut sink = sink.with(|msg: ServerMessage| {
         ready(
@@ -68,17 +68,23 @@ async fn handle_socket(websocket: WebSocket) -> Result<()> {
 
     greet(&mut sink).await?;
 
+    let mut source = source.map_err(Into::into).try_filter_map(|msg| {
+        ready(match msg.to_str() {
+            Ok(json) => from_str::<ClientMessage>(json)
+                .context("Parsing JSON")
+                .map(Some),
+            Err(()) => {
+                debug!("Non-text message");
+                Ok(None)
+            }
+        })
+    });
+
     let input_task = async {
         loop {
             match source.next().await {
-                Some(Ok(msg)) => match msg.to_str() {
-                    Ok(json) => {
-                        let msg = from_str::<ClientMessage>(json).context("Parsing JSON")?;
-                        debug!("Client message: {:?}", &msg);
-                    }
-                    Err(()) => debug!("Non-text message"),
-                },
-                Some(Err(error)) => return Err(error.into()),
+                Some(Ok(msg)) => debug!("Client message: {:?}", &msg),
+                Some(Err(error)) => return Err(error),
                 None => break Ok(()),
             }
         }