Lewei: send the stop command when the slot stops (#269)

CHANGELOG.md

@@ -12,6 +12,12 @@ Versions after 2.17 are `MAJOR.MINOR.PATCH` (2.17 is followed by 2.17.1 or 2.18.
 `docs/testing.md`, "Versions and releases"); the headings below keep the numbers as they
 were released.
 
+## 2.21.1 — 2026-11-07
+
+- A Lewei toy (`lewei://`) now gets the stop command (3) when the slot stops, on the same
+  connection and before it closes, as the vendor app does. Before, the command was never sent
+  and the toy saw only the closed connection (#269).
+
 ## 2.21.0 — 2026-11-05
 
 - Toy drones with a Lewei camera (vendor app WiFi UFO): the new slot URL `lewei://192.168.0.1`

android/app/build.gradle

@@ -23,8 +23,8 @@ android {
         applicationId "com.openipc.decoder"
         minSdk 21
         targetSdk 36
-        versionCode 52
-        versionName "2.21.0"
+        versionCode 53
+        versionName "2.21.1"
 
         buildConfigField "String", "GIT_HASH", "\"${getGitHash()}\""
     }

android/app/src/main/java/com/openipc/decoder/Lewei.java

@@ -59,6 +59,8 @@ final class Lewei {
     static final long QUIET_MS = 5000;
     private static final int CONNECT_TIMEOUT_MS = 5000;
     private static final int READ_TICK_MS = 200;
+    /** How long close() lets the session send the stop command before it closes the TCP. */
+    static volatile long closeGraceMs = 1000;
     /** Messages logged whole from the start of a session, for reports from the field. */
     private static final int LOGGED_MESSAGES = 6;
 
@@ -382,15 +384,23 @@ interface Listener {
     /**
      * One session on the calling thread until {@code keepGoing} fails or close(). A lost
      * connection or a toy that stopped sending ends run() with an exception: the session
-     * starts again.
+     * starts again. Once the stream was asked for, the session ends with the stop command on
+     * the same connection, as the vendor app and pylwdrone do (#269).
      */
     static final class Client implements Closeable {
         private final String tag;
         private final Url url;
         private final Jieli.TcpSetup setup;
         private final Socket socket = new Socket();
         private volatile boolean closed;
-        /** The stream was asked for on this connection. */
+        private final java.util.concurrent.CountDownLatch ended =
+                new java.util.concurrent.CountDownLatch(1);
+        private final java.util.concurrent.atomic.AtomicBoolean closerArmed =
+                new java.util.concurrent.atomic.AtomicBoolean();
+        /**
+         * The stream was asked for on this connection: from now on close() leaves the socket
+         * to run(), which sends the stop first.
+         */
         volatile boolean streamRequested;
 
         Client(String tag, Url url, Jieli.TcpSetup setup) {
@@ -400,6 +410,29 @@ static final class Client implements Closeable {
         }
 
         void run(KeepGoing keepGoing, Listener listener) throws IOException {
+            try {
+                session(keepGoing, listener);
+            } finally {
+                // A write blocked on a toy that stopped reading ends when the socket is closed
+                // after the grace period, on every way out, not only after close() (Qodo on #270)
+                if (streamRequested && !socket.isClosed()) {
+                    armCloser();
+                    try {
+                        socket.getOutputStream().write(command(CMD_STOP));
+                    } catch (IOException ignored) {
+                        // The connection is gone: the toy sees the TCP close
+                    }
+                }
+                try {
+                    socket.close();
+                } catch (IOException ignored) {
+                    // Closed is closed
+                }
+                ended.countDown();
+            }
+        }
+
+        private void session(KeepGoing keepGoing, Listener listener) throws IOException {
             InetAddress target = InetAddress.getByName(url.host);
             String bound = setup.prepare(socket, target);
             listener.onState(false, "Waiting for the toy: connecting to " + url.endpoint());
@@ -414,6 +447,9 @@ void run(KeepGoing keepGoing, Listener listener) throws IOException {
             socket.setSoTimeout(READ_TICK_MS);
             InputStream in = socket.getInputStream();
             OutputStream out = socket.getOutputStream();
+            // Stopped before the request: no request. close() closes the socket until the flag
+            // below is set, so a stop during this write ends it (Qodo on #270)
+            if (closed) return;
             out.write(command(CMD_START));
             streamRequested = true;
             listener.onState(false, "Waiting for the toy: asked for video on " + url.endpoint());
@@ -473,7 +509,6 @@ void run(KeepGoing keepGoing, Listener listener) throws IOException {
                                 + QUIET_MS / 1000 + " s");
                     }
                 }
-                out.write(command(CMD_STOP));
             } catch (IOException e) {
                 if (!closed && keepGoing.check()) throw e;
             }
@@ -494,14 +529,41 @@ private static long nowMs() {
             return System.nanoTime() / 1_000_000;
         }
 
+        /**
+         * Ends run() from any thread, the UI thread too: no socket write here. Before the stream
+         * was asked for the socket closes at once, which also ends a connect or the request
+         * itself; after that it is left to the session thread, which sends the stop command and
+         * closes it within one read tick (as Jieli does, #188). A stop that comes that late is
+         * the one risk: a new session may already have connected.
+         */
         @Override
         public void close() {
             closed = true;
-            try {
-                socket.close();
-            } catch (IOException e) {
-                // Closed is closed
+            if (!streamRequested) {
+                try {
+                    socket.close();
+                } catch (IOException e) {
+                    // Closed is closed
+                }
+                return;
             }
+            armCloser();
+        }
+
+        /** Closes the socket after closeGraceMs unless run() has ended by then; once. */
+        private void armCloser() {
+            if (ended.getCount() == 0 || !closerArmed.compareAndSet(false, true)) return;
+            Thread t = new Thread(() -> {
+                try {
+                    if (!ended.await(closeGraceMs, java.util.concurrent.TimeUnit.MILLISECONDS)) {
+                        socket.close();
+                    }
+                } catch (InterruptedException | IOException ignored) {
+                    // Nothing more to do
+                }
+            }, tag + "-lewei-close");
+            t.setDaemon(true);
+            t.start();
         }
     }
 }

android/app/src/test/java/com/openipc/decoder/LeweiTest.java

@@ -391,4 +391,126 @@ public boolean onRtp(byte[] buf, int offset, int length) {
             }
         }
     }
+
+    /**
+     * close() as RtspSession.stop() calls it, from another thread while video plays: the stop
+     * command still reaches the toy on the same connection, and run() ends quietly (#269).
+     */
+    @Test(timeout = 15000)
+    public void closeFromAnotherThreadSendsTheStop() throws Exception {
+        try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) {
+            AtomicInteger stops = new AtomicInteger();
+            CountDownLatch done = new CountDownLatch(1);
+            AtomicReference<Throwable> failed = new AtomicReference<>();
+            Thread toy = new Thread(() -> {
+                try (Socket s = server.accept()) {
+                    DataInputStream in = new DataInputStream(s.getInputStream());
+                    OutputStream out = s.getOutputStream();
+                    byte[] m = new byte[Lewei.HEADER];
+                    while (true) {
+                        in.readFully(m);
+                        if (m[10] == Lewei.CMD_START) {
+                            out.write(video(true, 1000, new byte[]{0, 0, 1, 0x27, 1}));
+                        } else if (m[10] == Lewei.CMD_STOP) {
+                            stops.incrementAndGet();
+                            // The connection closes after the stop, not before
+                            if (in.read() != -1) throw new IOException("data after the stop");
+                            return;
+                        }
+                    }
+                } catch (IOException | RuntimeException e) {
+                    failed.set(e);
+                } finally {
+                    done.countDown();
+                }
+            });
+            toy.setDaemon(true);
+            toy.start();
+
+            Lewei.Url u = Lewei.Url.parse("lewei://127.0.0.1:" + server.getLocalPort());
+            Lewei.Client c = new Lewei.Client("t", u, (sock, target) -> "default");
+            CountDownLatch receiving = new CountDownLatch(1);
+            AtomicReference<Throwable> runFailed = new AtomicReference<>();
+            CountDownLatch runEnded = new CountDownLatch(1);
+            Thread session = new Thread(() -> {
+                try {
+                    c.run(() -> true, new Lewei.Listener() {
+                        @Override
+                        public void onState(boolean on, String text) {
+                            if (on) receiving.countDown();
+                        }
+
+                        @Override
+                        public boolean onRtp(byte[] buf, int offset, int length) {
+                            return true;
+                        }
+                    });
+                } catch (Throwable e) {
+                    runFailed.set(e);
+                } finally {
+                    runEnded.countDown();
+                }
+            });
+            session.setDaemon(true);
+            session.start();
+            assertTrue(receiving.await(5, TimeUnit.SECONDS));
+            c.close();
+            assertTrue(runEnded.await(5, TimeUnit.SECONDS));
+            assertTrue(done.await(5, TimeUnit.SECONDS));
+            if (runFailed.get() != null) throw new AssertionError(runFailed.get());
+            if (failed.get() != null) throw new AssertionError(failed.get());
+            assertEquals(1, stops.get());
+        }
+    }
+
+    /** A toy that sends no video and reads nothing more: close() still ends run() (#269). */
+    @Test(timeout = 15000)
+    public void closeEndsASessionBeforeAnyVideo() throws Exception {
+        long grace = Lewei.closeGraceMs;
+        Lewei.closeGraceMs = 300;
+        try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) {
+            CountDownLatch hold = new CountDownLatch(1);
+            Thread toy = new Thread(() -> {
+                try (Socket s = server.accept()) {
+                    s.getInputStream().read(new byte[Lewei.HEADER]);
+                    hold.await();
+                } catch (IOException | InterruptedException e) {
+                    // the test sees it
+                }
+            });
+            toy.setDaemon(true);
+            toy.start();
+            Lewei.Url u = Lewei.Url.parse("lewei://127.0.0.1:" + server.getLocalPort());
+            Lewei.Client c = new Lewei.Client("t", u, (sock, target) -> "default");
+            CountDownLatch runEnded = new CountDownLatch(1);
+            Thread session = new Thread(() -> {
+                try {
+                    c.run(() -> true, new Lewei.Listener() {
+                        @Override
+                        public void onState(boolean on, String text) {
+                        }
+
+                        @Override
+                        public boolean onRtp(byte[] buf, int offset, int length) {
+                            return true;
+                        }
+                    });
+                } catch (IOException e) {
+                    // closed: no exception expected, but the end is what counts here
+                } finally {
+                    runEnded.countDown();
+                }
+            });
+            session.setDaemon(true);
+            session.start();
+            long until = System.nanoTime() + 5_000_000_000L;
+            while (!c.streamRequested && System.nanoTime() < until) Thread.sleep(10);
+            assertTrue(c.streamRequested);
+            c.close();
+            assertTrue(runEnded.await(5, TimeUnit.SECONDS));
+            hold.countDown();
+        } finally {
+            Lewei.closeGraceMs = grace;
+        }
+    }
 }

docs/toys.md

@@ -143,7 +143,7 @@ To use it: join the `HF_…` Wi‑Fi, [stay connected](#connecting), close the v
 What the app does, as a capture of the vendor app shows it; [pylwdrone](https://github.com/meekworth/pylwdrone) (Apache-2.0) names the fields:
 
 - It connects to TCP port 7060. Every message starts with a 46-byte header: `lewei_cmd`, a zero byte and nine little-endian 32-bit numbers (the command, an argument, one unused, the length of what follows, the stream type, two keys).
-- Command 2 with the argument 1 asks for the high-definition stream, and the video starts at once. Command 1 every second keeps it going, command 3 ends it.
+- Command 2 with the argument 1 asks for the high-definition stream, and the video starts at once. Command 1 every second keeps it going, command 3 ends it: the app sends it on the same connection when the slot stops or the session starts over, then closes the connection.
 - The video is **H.264** in Annex B form, one frame in each message of command 257, behind 32 bytes of its own (a flag, the length of the frame, a counter in ms). The tested drone sends 1280×720 at 20 fps, about 2.8 Mbit/s.
 - With stream type 1 the camera has **inverted one byte** of each frame, at a place that follows from the length and the counter; the app inverts it back. Without that, frames decode only partly. Type 129 is not handled: the log says so, and the picture will have errors.
 - If no video comes for 5 s, the app connects again.