diff --git a/CHANGELOG.md b/CHANGELOG.md index e6b4df0..1cd8b90 100644 --- a/CHANGELOG.md +++ b/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` diff --git a/android/app/build.gradle b/android/app/build.gradle index faf701b..7ad57d1 100755 --- a/android/app/build.gradle +++ b/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()}\"" } diff --git a/android/app/src/main/java/com/openipc/decoder/Lewei.java b/android/app/src/main/java/com/openipc/decoder/Lewei.java index 32e95cd..4230a3f 100644 --- a/android/app/src/main/java/com/openipc/decoder/Lewei.java +++ b/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,7 +384,8 @@ 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; @@ -390,7 +393,14 @@ static final class Client implements Closeable { 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(); } } } diff --git a/android/app/src/test/java/com/openipc/decoder/LeweiTest.java b/android/app/src/test/java/com/openipc/decoder/LeweiTest.java index 8628534..22d7620 100644 --- a/android/app/src/test/java/com/openipc/decoder/LeweiTest.java +++ b/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 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 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; + } + } } diff --git a/docs/toys.md b/docs/toys.md index 77a4798..fe4a7f7 100644 --- a/docs/toys.md +++ b/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.