core+tests: Socket testing and improvements

Cargo.toml

@@ -25,6 +25,8 @@ members = [
 
     "tests",
     "tests/input-format",
+    "tests/socket-format",
+    "tests/mocket",
 ]
 default-members = ["desktop"]
 resolver = "2"

core/src/backend/navigator.rs

@@ -289,7 +289,7 @@ impl NullExecutor {
     }
 
     pub fn run(&mut self) {
-        self.0.run();
+        self.0.run_until_stalled();
     }
 }
 

core/src/socket.rs

@@ -7,7 +7,6 @@ use gc_arena::Collect;
 use generational_arena::{Arena, Index};
 use std::{
     cell::RefCell,
-    collections::VecDeque,
     sync::mpsc::{channel, Receiver, Sender},
     time::Duration,
 };
@@ -19,15 +18,13 @@ pub type SocketHandle = Index;
 struct Socket<'gc> {
     target: SocketObject<'gc>,
     sender: RefCell<Sender<Vec<u8>>>,
-    send_buffer: VecDeque<Vec<u8>>,
 }
 
 impl<'gc> Socket<'gc> {
     fn new(target: SocketObject<'gc>, sender: Sender<Vec<u8>>) -> Self {
         Self {
             target,
             sender: RefCell::new(sender),
-            send_buffer: Default::default(),
         }
     }
 }
@@ -107,8 +104,8 @@ impl<'gc> Sockets<'gc> {
     }
 
     pub fn send(&mut self, handle: SocketHandle, data: Vec<u8>) {
-        if let Some(Socket { send_buffer, .. }) = self.sockets.get_mut(handle) {
-            send_buffer.push_back(data);
+        if let Some(Socket { sender, .. }) = self.sockets.get_mut(handle) {
+            let _ = sender.borrow().send(data);
         }
     }
 
@@ -210,17 +207,5 @@ impl<'gc> Sockets<'gc> {
                 }
             }
         }
-
-        for (_handle, socket) in context.sockets.sockets.iter_mut() {
-            let Socket {
-                sender,
-                send_buffer,
-                ..
-            } = socket;
-
-            if let Some(to_send) = send_buffer.pop_front() {
-                let _ = sender.borrow().send(to_send);
-            }
-        }
     }
 }

tests/Cargo.toml

@@ -13,6 +13,7 @@ ruffle_core = { path = "../core", features = ["deterministic", "timeline_debug",
 ruffle_render_wgpu = { path = "../render/wgpu" }
 ruffle_render = { path = "../render" }
 ruffle_input_format = { path = "input-format" }
+ruffle_socket_format = { path = "socket-format" }
 ruffle_video_software = { path = "../video/software", optional = true }
 image = { version = "0.24.5", default-features = false, features = ["png"] }
 regex = "1.7.1"
@@ -35,6 +36,7 @@ toml = "0.7.4"
 libtest-mimic = "0.6.0"
 walkdir = "2.3.2"
 anyhow = "1.0"
+async-io = "1.13.0"
 
 [[test]]
 name = "tests"

tests/mocket/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "mocket"
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+repository.workspace = true
+version.workspace = true
+
+[dependencies]
+anyhow = "1.0"
+clap = { version = "4.3.12", features = ["derive"] }
+tracing = { workspace = true}
+tracing-subscriber = { workspace = true }
+ruffle_socket_format = { path = "../socket-format" }

tests/mocket/src/main.rs

@@ -0,0 +1,127 @@
+use anyhow::Error;
+use clap::Parser;
+use ruffle_socket_format::SocketEvent;
+use std::{
+    io::{Read, Write},
+    net::TcpListener,
+    path::PathBuf,
+};
+use tracing_subscriber::filter::{EnvFilter, LevelFilter};
+
+static POLICY: &[u8] = b"<?xml version=\"1.0\"?>
+<!DOCTYPE cross-domain-policy SYSTEM \"http://www.adobe.com/xml/dtds/cross-domain-policy.dtd\">
+<cross-domain-policy>
+<allow-access-from domain=\"*\" to-ports=\"*\"/>
+</cross-domain-policy>\0";
+
+#[derive(Parser, Debug)]
+struct Opt {
+    /// Path to a `socket.json` file.
+    #[clap(name = "FILE")]
+    file_path: PathBuf,
+}
+
+fn main() -> Result<(), Error> {
+    let opt = Opt::parse();
+
+    let subscriber = tracing_subscriber::fmt::Subscriber::builder()
+        .with_env_filter(
+            EnvFilter::builder()
+                .with_default_directive(LevelFilter::INFO.into())
+                .from_env_lossy(),
+        )
+        .finish();
+    // Ignore error if it's already been set
+    let _ = tracing::subscriber::set_global_default(subscriber);
+
+    let events = SocketEvent::from_file(opt.file_path)?;
+    let event_count = events.len();
+
+    let listener = TcpListener::bind("0.0.0.0:8001")?;
+    tracing::info!("Listening on {}", listener.local_addr()?);
+    let (mut stream, addr) = listener.accept()?;
+    tracing::info!("Incoming connection from {}", addr);
+
+    // Handle socket policy stuff. (Required as Flash Player wont want to connect otherwise.)
+    let mut buffer = [0; 4096];
+    let _ = stream.read(&mut buffer);
+    stream.write_all(POLICY)?;
+    tracing::info!("Policy sent successfully!");
+
+    // Now we listen again as flash reopens socket connection.
+    let (mut stream, addr) = listener.accept()?;
+    tracing::info!("Incoming connection from {}", addr);
+
+    for (index, event) in events.into_iter().enumerate() {
+        tracing::info!("Running step {}/{}", index + 1, event_count);
+
+        match event {
+            SocketEvent::Receive { expected } => {
+                let mut output = vec![];
+
+                loop {
+                    let mut buffer = [0; 4096];
+
+                    match stream.read(&mut buffer) {
+                        Err(_) | Ok(0) => {
+                            tracing::error!("Expected data, but socket was closed.");
+                            return Ok(());
+                        }
+                        Ok(read) => {
+                            if read == 4096 {
+                                output.extend(buffer);
+                            } else {
+                                let data = buffer.into_iter().take(read).collect::<Vec<_>>();
+                                output.extend(data);
+                                break;
+                            }
+                        }
+                    }
+                }
+
+                if output != expected {
+                    tracing::error!(
+                        "Received data did not match expected data\nExpected: {:?}\nActual: {:?}",
+                        expected,
+                        output
+                    );
+                }
+            }
+            SocketEvent::Send { mut payload } => {
+                while !payload.is_empty() {
+                    match stream.write(&payload) {
+                        Err(_) | Ok(0) => {
+                            tracing::error!("Socket was closed in middle of writing.");
+                            return Ok(());
+                        }
+                        Ok(written) => {
+                            let _ = payload.drain(..written);
+                        }
+                    }
+                }
+            }
+            SocketEvent::WaitForDisconnect => {
+                let mut buffer = [0; 4096];
+
+                match stream.read(&mut buffer) {
+                    Err(_) | Ok(0) => {
+                        tracing::info!("Client has closed the connection!");
+                        return Ok(());
+                    }
+                    Ok(_) => {
+                        tracing::error!(
+                            "Expected client to close connection, but data was sent instead."
+                        );
+                    }
+                }
+            }
+            SocketEvent::Disconnect => {
+                tracing::info!("Disconnecting client.");
+                drop(stream);
+                break;
+            }
+        }
+    }
+
+    Ok(())
+}

tests/socket-format/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "ruffle_socket_format"
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+repository.workspace = true
+version.workspace = true
+
+[dependencies]
+serde = { version = "1.0.152", features = ["derive"] }
+serde_json = "1.0.91"

tests/socket-format/src/lib.rs

@@ -0,0 +1,28 @@
+use std::{fs::File, io, path::Path};
+
+use serde::{Deserialize, Serialize};
+use serde_json::from_reader;
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum SocketEvent {
+    /// Wait for input data that matches this.
+    Receive { expected: Vec<u8> },
+    /// Send data to client.
+    Send { payload: Vec<u8> },
+    /// Expect client to disconnect.
+    WaitForDisconnect,
+    /// Disconnect the client.
+    Disconnect,
+}
+
+impl SocketEvent {
+    pub fn from_file<P>(path: P) -> Result<Vec<Self>, io::Error>
+    where
+        P: AsRef<Path>,
+    {
+        let file = File::open(path)?;
+
+        Ok(from_reader(file)?)
+    }
+}

tests/tests/swfs/avm2/socket_close/Test.as

@@ -0,0 +1,25 @@
+package
+{
+    import flash.display.Sprite;
+
+    public class Test extends Sprite
+    {
+    }
+}
+
+import flash.net.Socket;
+import flash.events.Event;
+
+var socket:Socket = new Socket();
+
+socket.addEventListener(Event.CONNECT, function(evt:Event):void
+{
+    trace("connected");
+});
+
+socket.addEventListener(Event.CLOSE, function(evt:Event):void
+{
+    trace("closed");
+});
+
+socket.connect("localhost", 8001);

tests/tests/swfs/avm2/socket_close/output.txt

@@ -0,0 +1,2 @@
+connected
+closed

tests/tests/swfs/avm2/socket_close/socket.json

@@ -0,0 +1,5 @@
+[
+    {
+        "type": "Disconnect"
+    }
+]

tests/tests/swfs/avm2/socket_close/test.toml

@@ -0,0 +1 @@
+num_ticks = 10

tests/tests/swfs/avm2/socket_connect/Test.as

@@ -0,0 +1,24 @@
+package
+{
+    import flash.display.Sprite;
+
+    public class Test extends Sprite
+    {
+    }
+}
+
+import flash.events.Event;
+import flash.net.Socket;
+
+var socket:Socket = new Socket();
+
+socket.addEventListener(Event.CONNECT, function(event:Event):void
+{
+    trace("connected");
+    socket.writeUTF("Hello!");
+    socket.flush();
+    socket.close();
+});
+
+socket.connect("localhost", 8001);
+

tests/tests/swfs/avm2/socket_connect/output.txt

@@ -0,0 +1 @@
+connected