diff --git a/android/app/src/main/java/com/openipc/decoder/Decoder.java b/android/app/src/main/java/com/openipc/decoder/Decoder.java index d00efd27..b592638c 100755 --- a/android/app/src/main/java/com/openipc/decoder/Decoder.java +++ b/android/app/src/main/java/com/openipc/decoder/Decoder.java @@ -1558,6 +1558,8 @@ private Map debugState() { // The toy drone's protocol counters (#187) Map ufo = rtspClient != null ? rtspClient.flowUfoStats() : null; if (ufo != null) single.put("flowufo", ufo); + Map jl = rtspClient != null ? rtspClient.jieliStats() : null; + if (jl != null) single.put("jieli", jl); Map hb = rtspClient != null ? rtspClient.heartbeatStats() : null; if (hb != null) single.put("heartbeat", hb); m.put("single", single); @@ -1600,6 +1602,8 @@ private Map debugState() { c.put("audio", i == quadAudio); Map ufo = cell.flowUfoStats(); if (ufo != null) c.put("flowufo", ufo); + Map jl = cell.jieliStats(); + if (jl != null) c.put("jieli", jl); Map hb = cell.heartbeatStats(); if (hb != null) c.put("heartbeat", hb); } @@ -1670,6 +1674,7 @@ private String slotTransport(int i, String url) { if (RtpListener.Url.matches(url)) return "UDP listen"; if (HttpMjpeg.matches(url)) return "HTTP"; if (FlowUfo.Url.matches(url)) return "FLOW-UFO UDP"; + if (Jieli.Url.matches(url)) return "Jieli RTP"; Transport.Mode m = StreamProbe.probeMode(mTransports[i]); return "check over " + (m == Transport.Mode.MULTICAST ? "multicast" : m.name()); } diff --git a/android/app/src/main/java/com/openipc/decoder/FlowUfo.java b/android/app/src/main/java/com/openipc/decoder/FlowUfo.java index 594580e9..62e67eda 100644 --- a/android/app/src/main/java/com/openipc/decoder/FlowUfo.java +++ b/android/app/src/main/java/com/openipc/decoder/FlowUfo.java @@ -635,19 +635,19 @@ static String subnetHint(InetAddress target, List local) { + " flowufo://192.168.169.1"); } if (on80 && !on169 && !on1) { - sentence(parts, "Toys on 192.168.80.x (HF_ Wi-Fi) use another protocol, not supported" - + " yet"); + sentence(parts, "This phone is on 192.168.80.x (HF_ Wi-Fi), a Jieli toy: try" + + " jieli://192.168.80.1"); } return parts.length() > 0 ? parts.toString() : null; } - private static void sentence(StringBuilder sb, String s) { + static void sentence(StringBuilder sb, String s) { if (sb.length() > 0) sb.append(". "); sb.append(s); } /** {@code a} is in 192.168.{@code third}.0/24. */ - private static boolean in24(InetAddress a, int third) { + static boolean in24(InetAddress a, int third) { if (!(a instanceof Inet4Address)) return false; byte[] b = a.getAddress(); return (b[0] & 0xFF) == 192 && (b[1] & 0xFF) == 168 && (b[2] & 0xFF) == third; diff --git a/android/app/src/main/java/com/openipc/decoder/Jieli.java b/android/app/src/main/java/com/openipc/decoder/Jieli.java new file mode 100644 index 00000000..4fa128ba --- /dev/null +++ b/android/app/src/main/java/com/openipc/decoder/Jieli.java @@ -0,0 +1,1126 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * Jieli.java — Jieli toy drones ("HF_xxxxxxxx" Wi-Fi): CTP over TCP, RTP to UDP 6666 (#187) + * + */ + +package com.openipc.decoder; + +import android.util.Log; + +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.Inet4Address; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; +import java.text.SimpleDateFormat; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +/** + * Toy drones with an "HF_xxxxxxxx" access point (Jieli AC79 camera, vendor app "DVRunning 2") + * at 192.168.80.1 (#187, #196). The app talks CTP to TCP port 3333: "CTP:", the topic length + * (u16 LE), the topic, the JSON length (u32 LE) and the JSON, which may be empty. The session + * as a PCAPdroid capture of the vendor app shows it: APP_ACCESS; the toy announces + * KEEP_ALIVE_INTERVAL, VIDEO_PARAM and more; DATE_TIME; CTP_KEEP_ALIVE every 7.5 s; + * OPEN_RT_STREAM; CLOSE_RT_STREAM at the end. The video itself was not in the capture: after + * turbodrone's research on the same SDK (marshallrichards/turbodrone, Apache-2.0, used as a + * description only) it is RTP/JPEG (payload type 26) or H.264 (96) to the phone's UDP port + * 6666. The packets go into the app's normal RTP paths. Unverified, so the log shows what + * really arrives on 6666, and on the audio and stream ports 1234 and 2224 too when 6666 stays + * silent. Video only: FLIGHT_CONTROL is never sent, nothing goes to the toy's UDP port 3456. + * Pure Java apart from the log. + */ +final class Jieli { + private static final String TAG = "OpenIPCDecoder"; + + static final String SCHEME = "jieli://"; + static final String DEFAULT_HOST = "192.168.80.1"; + static final int CTP_PORT = 3333; + static final int RTP_PORT = 6666; + static final int DEFAULT_WIDTH = 640; + static final int DEFAULT_HEIGHT = 480; + static final int DEFAULT_FPS = 25; + static final int FORMAT_JPEG = 0; + static final int FORMAT_H264 = 1; + static final int PT_JPEG = 26; + static final int PT_H264 = 96; + + static final byte[] MAGIC = {'C', 'T', 'P', ':'}; + /** Longer topics or JSON than this are not CTP: the reader looks for the next "CTP:". */ + static final int MAX_TOPIC = 256; + static final int MAX_JSON = 256 * 1024; + + static final String APP_ACCESS = "APP_ACCESS"; + static final String DATE_TIME = "DATE_TIME"; + static final String KEEP_ALIVE = "CTP_KEEP_ALIVE"; + static final String KEEP_ALIVE_INTERVAL = "KEEP_ALIVE_INTERVAL"; + static final String VIDEO_PARAM = "VIDEO_PARAM"; + static final String OPEN_RT_STREAM = "OPEN_RT_STREAM"; + static final String CLOSE_RT_STREAM = "CLOSE_RT_STREAM"; + + private static final int CONNECT_TIMEOUT_MS = 5000; + /** Read timeout of the CTP socket: how often the timers are looked at. */ + private static final int TICK_MS = 100; + /** The keepalive of the vendor app; a shorter announced timeout makes it timeout / 4. */ + static final long KEEP_ALIVE_MAX_MS = 7500; + private static final long KEEP_ALIVE_MIN_MS = 250; + // Not final so unit tests can shorten them + /** How long APP_ACCESS waits for VIDEO_PARAM before the stream is asked for anyway. */ + static volatile long paramWaitMs = 1500; + /** No video this long after OPEN_RT_STREAM: said in the log and status, diagnostics on. */ + static volatile long noVideoMs = 5000; + /** No video this long (after a request or the last packet): OPEN_RT_STREAM again. */ + static volatile long reopenMs = 10_000; + /** Video that stops for this long is "quiet". */ + static volatile long quietMs = 3000; + /** The toy says nothing on CTP for this long (at least): the connection is given up. */ + static volatile long ctpSilenceMs = 20_000; + /** + * Ports opened only to log what arrives when 6666 stays silent: 1234 (audio in the + * vendor SDK) and 2224 (its RT stream port). Nothing received there is played. + */ + static volatile int[] diagnosticPorts = {1234, 2224}; + + private Jieli() {} + + // ====================== URL ====================== + + /** + * jieli://[host][:port][/...][?option=value&...]: the toy's CTP address, 192.168.80.1:3333 + * when left out. Options, as in udp:// URLs (unknown ones are ignored, bad values + * refused): w, h, fps (the stream asked for; default what the toy announces in + * VIDEO_PARAM, else 640x480 at 25), rtp (the local UDP port for the video, 6666), + * format (0 JPEG, the default, or 1 H.264). + */ + static final class Url { + String host = DEFAULT_HOST; + int port = CTP_PORT; + int rtpPort = RTP_PORT; + /** 0: from VIDEO_PARAM, else the default. */ + int width; + int height; + int fps; + int format = FORMAT_JPEG; + + static boolean matches(String url) { + return url != null && url.trim().toLowerCase(Locale.ROOT).startsWith(SCHEME); + } + + static Url parse(String url) throws IOException { + if (!matches(url)) throw new IOException("Not a jieli:// URL"); + String rest = url.trim().substring(SCHEME.length()); + int end = rest.length(); + for (int i = 0; i < rest.length(); i++) { + char ch = rest.charAt(i); + if (ch == '/' || ch == '?' || ch == '#') { + end = i; + break; + } + } + String authority = rest.substring(0, end).trim(); + String query = ""; + int q = rest.indexOf('?'); + int hash = rest.indexOf('#'); + if (q >= 0 && (hash < 0 || q < hash)) { + query = rest.substring(q + 1, hash >= 0 ? hash : rest.length()); + } + Url u = new Url(); + String host = authority; + int colon = authority.lastIndexOf(':'); + if (authority.startsWith("[")) { + int close = authority.indexOf(']'); + if (close < 0) throw new IOException("Bad host in URL"); + host = authority.substring(1, close); + colon = authority.indexOf(':', close); + } else if (colon >= 0) { + host = authority.substring(0, colon); + } + if (colon >= 0) u.port = number(authority.substring(colon + 1), 1, 65535, "port"); + if (!host.isEmpty()) u.host = host; + for (int i = 0; i < u.host.length(); i++) { + char ch = u.host.charAt(i); + if (ch <= 0x20 || ch == 0x7F || ch == '@') throw new IOException("Bad host in URL"); + } + for (String param : query.split("&")) { + int eq = param.indexOf('='); + if (eq < 0) continue; + String name = param.substring(0, eq).trim().toLowerCase(Locale.ROOT); + String value = param.substring(eq + 1).trim(); + switch (name) { + case "w": + u.width = number(value, 16, 4096, "w"); + break; + case "h": + u.height = number(value, 16, 4096, "h"); + break; + case "fps": + u.fps = number(value, 1, 120, "fps"); + break; + case "rtp": + u.rtpPort = number(value, 1, 65535, "rtp"); + break; + case "format": + u.format = number(value, 0, 1, "format (0 JPEG, 1 H.264)"); + break; + default: + break; + } + } + return u; + } + + private static int number(String s, int min, int max, String what) throws IOException { + int v; + try { + v = Integer.parseInt(s.trim()); + } catch (NumberFormatException e) { + throw new IOException("Bad " + what + " in URL: " + s); + } + if (v < min || v > max) throw new IOException("Bad " + what + " in URL: " + s); + return v; + } + + String endpoint() { + return (host.indexOf(':') >= 0 ? "[" + host + "]" : host) + ":" + port; + } + } + + // ====================== CTP messages ====================== + + /** One CTP message: a topic and its JSON ("" when there is none). */ + static final class Message { + final String topic; + final String json; + + Message(String topic, String json) { + this.topic = topic; + this.json = json; + } + + @Override + public String toString() { + return json.isEmpty() ? topic : topic + " " + json; + } + } + + /** "CTP:" + topic length (u16 LE) + topic + JSON length (u32 LE) + JSON. */ + static byte[] encode(String topic, String json) { + byte[] t = topic.getBytes(StandardCharsets.UTF_8); + byte[] j = json != null ? json.getBytes(StandardCharsets.UTF_8) : new byte[0]; + byte[] out = new byte[MAGIC.length + 2 + t.length + 4 + j.length]; + System.arraycopy(MAGIC, 0, out, 0, MAGIC.length); + int pos = MAGIC.length; + out[pos] = (byte) t.length; + out[pos + 1] = (byte) (t.length >> 8); + pos += 2; + System.arraycopy(t, 0, out, pos, t.length); + pos += t.length; + for (int i = 0; i < 4; i++) out[pos + i] = (byte) (j.length >> (8 * i)); + pos += 4; + System.arraycopy(j, 0, out, pos, j.length); + return out; + } + + /** + * Cuts the TCP byte stream into messages: a message may arrive in pieces or several in + * one read. Bytes that are not a message (no "CTP:", a length out of range) are skipped + * up to the next "CTP:" and counted. + */ + static final class Reader { + private byte[] buf = new byte[4096]; + private int start; + private int end; + long skipped; + + void feed(byte[] b, int off, int len) { + if (end + len > buf.length) { + int have = end - start; + if (have + len > buf.length) { + byte[] n = new byte[Math.max(buf.length * 2, have + len)]; + System.arraycopy(buf, start, n, 0, have); + buf = n; + } else { + System.arraycopy(buf, start, buf, 0, have); + } + start = 0; + end = have; + } + System.arraycopy(b, off, buf, end, len); + end += len; + } + + /** The next whole message, or null until more bytes arrive. */ + Message next() { + while (true) { + int have = end - start; + if (have < MAGIC.length) return null; + if (!startsWithMagic(start)) { + skipToMagic(); + continue; + } + if (have < MAGIC.length + 2) return null; + int topicLen = (buf[start + 4] & 0xFF) | (buf[start + 5] & 0xFF) << 8; + if (topicLen > MAX_TOPIC) { + skipOne(); + continue; + } + int jsonAt = start + MAGIC.length + 2 + topicLen; + if (end < jsonAt + 4) return null; + long jsonLen = (buf[jsonAt] & 0xFFL) | (buf[jsonAt + 1] & 0xFFL) << 8 + | (buf[jsonAt + 2] & 0xFFL) << 16 | (buf[jsonAt + 3] & 0xFFL) << 24; + if (jsonLen > MAX_JSON) { + skipOne(); + continue; + } + int total = MAGIC.length + 2 + topicLen + 4 + (int) jsonLen; + if (have < total) return null; + String topic = new String(buf, start + 6, topicLen, StandardCharsets.UTF_8); + String json = new String(buf, jsonAt + 4, (int) jsonLen, StandardCharsets.UTF_8); + start += total; + if (start == end) start = end = 0; + return new Message(topic, json); + } + } + + private boolean startsWithMagic(int at) { + for (int i = 0; i < MAGIC.length; i++) { + if (buf[at + i] != MAGIC[i]) return false; + } + return true; + } + + /** Drops the first byte (a "CTP:" with bad lengths) and looks for the next one. */ + private void skipOne() { + start++; + skipped++; + skipToMagic(); + } + + private void skipToMagic() { + int i = start; + while (i + MAGIC.length <= end && !startsWithMagic(i)) i++; + if (i + MAGIC.length > end) { + // Keep a tail that may be the start of "CTP:" + i = Math.max(start, end - (MAGIC.length - 1)); + } + skipped += i - start; + start = i; + } + } + + // ====================== JSON ====================== + + /** {"op":"PUT"}, or with "param" from name/value pairs, in their order. */ + static String put(String... nameValues) { + StringBuilder sb = new StringBuilder("{\"op\":\"PUT\""); + if (nameValues.length > 0) { + sb.append(",\"param\":{"); + for (int i = 0; i + 1 < nameValues.length; i += 2) { + if (i > 0) sb.append(','); + quote(sb, nameValues[i]); + sb.append(':'); + quote(sb, nameValues[i + 1]); + } + sb.append('}'); + } + return sb.append('}').toString(); + } + + private static void quote(StringBuilder sb, String s) { + sb.append('"'); + for (int i = 0; i < s.length(); i++) { + char c = s.charAt(i); + if (c == '"' || c == '\\') { + sb.append('\\').append(c); + } else if (c < 0x20) { + sb.append(String.format(Locale.ROOT, "\\u%04x", (int) c)); + } else { + sb.append(c); + } + } + sb.append('"'); + } + + /** + * A small JSON reader (org.json is a stub in unit tests): objects become maps, arrays + * lists, strings strings, and numbers, true, false and null their text. Null when the + * text is not one JSON object. + */ + static Map parseJson(String s) { + if (s == null) return null; + try { + Json j = new Json(s); + j.space(); + if (j.peek() != '{') return null; + @SuppressWarnings("unchecked") + Map m = (Map) j.value(0); + j.space(); + return j.pos == s.length() ? m : null; + } catch (RuntimeException e) { + return null; + } + } + + private static final class Json { + final String s; + int pos; + + Json(String s) { + this.s = s; + } + + char peek() { + if (pos >= s.length()) throw new IllegalArgumentException("end"); + return s.charAt(pos); + } + + void space() { + while (pos < s.length() && Character.isWhitespace(s.charAt(pos))) pos++; + } + + void expect(char c) { + space(); + if (peek() != c) throw new IllegalArgumentException("expected " + c); + pos++; + } + + Object value(int depth) { + if (depth > 16) throw new IllegalArgumentException("too deep"); + space(); + char c = peek(); + if (c == '{') { + pos++; + Map m = new LinkedHashMap<>(); + space(); + if (peek() == '}') { + pos++; + return m; + } + while (true) { + space(); + String key = string(); + expect(':'); + m.put(key, value(depth + 1)); + space(); + if (peek() == ',') { + pos++; + continue; + } + expect('}'); + return m; + } + } + if (c == '[') { + pos++; + List l = new ArrayList<>(); + space(); + if (peek() == ']') { + pos++; + return l; + } + while (true) { + l.add(value(depth + 1)); + space(); + if (peek() == ',') { + pos++; + continue; + } + expect(']'); + return l; + } + } + if (c == '"') return string(); + int from = pos; + while (pos < s.length() && ",}] \t\r\n".indexOf(s.charAt(pos)) < 0) pos++; + if (pos == from) throw new IllegalArgumentException("value"); + return s.substring(from, pos); + } + + String string() { + if (peek() != '"') throw new IllegalArgumentException("string"); + pos++; + StringBuilder sb = new StringBuilder(); + while (true) { + char c = peek(); + pos++; + if (c == '"') return sb.toString(); + if (c != '\\') { + sb.append(c); + continue; + } + char e = peek(); + pos++; + switch (e) { + case 'n': sb.append('\n'); break; + case 't': sb.append('\t'); break; + case 'r': sb.append('\r'); break; + case 'b': sb.append('\b'); break; + case 'f': sb.append('\f'); break; + case 'u': + sb.append((char) Integer.parseInt(s.substring(pos, pos + 4), 16)); + pos += 4; + break; + default: sb.append(e); break; + } + } + } + } + + /** {@code name} in the message's "param" object, else at the top, as text; or null. */ + static String param(Map json, String name) { + if (json == null) return null; + Object p = json.get("param"); + if (p instanceof Map) { + Object v = ((Map) p).get(name); + if (v instanceof String) return (String) v; + } + Object v = json.get(name); + return v instanceof String ? (String) v : null; + } + + /** The integer {@code name} of a message, or -1. */ + static int intParam(Map json, String name) { + String v = param(json, name); + if (v == null) return -1; + try { + return Integer.parseInt(v.trim()); + } catch (NumberFormatException e) { + return -1; + } + } + + /** min(timeout / 4, 7.5 s), not below 250 ms; 7.5 s when the toy announced none. */ + static long keepAliveMs(long timeoutMs) { + if (timeoutMs <= 0) return KEEP_ALIVE_MAX_MS; + return Math.max(KEEP_ALIVE_MIN_MS, Math.min(timeoutMs / 4, KEEP_ALIVE_MAX_MS)); + } + + /** The request for the stream, in the field order of the vendor app. */ + static String openStream(int format, int width, int height, int fps) { + return put("format", String.valueOf(format), "h", String.valueOf(height), + "w", String.valueOf(width), "fps", String.valueOf(fps)); + } + + static String truncate(String s, int max) { + return s.length() <= max ? s : s.substring(0, max) + "…"; + } + + // ====================== Subnet hint ====================== + + /** + * A hint when a jieli:// slot cannot be right for the phone's networks, or null: the + * target is the phone itself, or the phone is on another toy family's network. + */ + static String subnetHint(InetAddress target, List local) { + if (!(target instanceof Inet4Address) || local == null) return null; + boolean own = local.contains(target); + boolean on1 = false; + boolean on169 = false; + boolean on80 = false; + for (InetAddress a : local) { + on1 |= FlowUfo.in24(a, 1); + on169 |= FlowUfo.in24(a, 169); + on80 |= FlowUfo.in24(a, 80); + } + StringBuilder parts = new StringBuilder(); + byte[] b = target.getAddress(); + if (own) { + FlowUfo.sentence(parts, target.getHostAddress() + " is this phone; the toy is usually " + + (b[0] & 0xFF) + "." + (b[1] & 0xFF) + "." + (b[2] & 0xFF) + ".1"); + } + if (on80 && !FlowUfo.in24(target, 80)) { + FlowUfo.sentence(parts, "This phone is on 192.168.80.x: the toy is usually" + + " jieli://192.168.80.1"); + } + if (!on80 && on169) { + FlowUfo.sentence(parts, "This network looks like a FLOW-UFO toy: try" + + " flowufo://192.168.169.1"); + } + if (!on80 && on1) { + FlowUfo.sentence(parts, "This network looks like a KY UFO RTSP toy: try" + + " rtsp://192.168.1.1:7070/webcam"); + } + return parts.length() > 0 ? parts.toString() : null; + } + + // ====================== Client ====================== + + interface Listener { + /** Waiting (no video) or receiving; text for the status. */ + void onState(boolean receiving, String text); + + /** One RTP packet with payload type 26 (JPEG) or 96 (H.264), valid during the call. */ + void onRtp(byte[] buf, int offset, int length); + + /** One whole JPEG in a datagram without RTP, valid during the call. */ + default void onJpeg(byte[] buf, int offset, int length) {} + } + + /** Binds the CTP socket to the toy's Wi-Fi before it connects; returns what was done. */ + interface TcpSetup { + String prepare(Socket socket, InetAddress target); + } + + /** + * One CTP session on the calling thread (connect, the commands, keepalives, the timers), + * and the video socket on a thread of its own, until {@code keepGoing} fails or close(). + * The stream is closed with CLOSE_RT_STREAM on the way out. A lost connection ends run() + * with an exception: the session starts again. + */ + static final class Client implements Closeable { + private final String tag; + private final Url url; + private final TcpSetup tcpSetup; + private final FlowUfo.SocketSetup udpSetup; + private volatile boolean closed; + /** The phone's addresses for the subnet hint; null: the real ones. For tests. */ + List localForHint; + + private final Object lock = new Object(); + private Socket tcp; + private boolean tcpConnected; + private final List udpSockets = new ArrayList<>(); + private volatile InetAddress address; + + // For the stats + private volatile String state = "connecting"; + private volatile String hint; + private volatile String networkTcp = "default"; + private volatile String networkUdp = "default"; + private volatile int ctpSent; + private volatile int ctpReceived; + private volatile int keepAlives; + private volatile int keepAliveReplies; + private volatile long keepAliveIntervalMs = KEEP_ALIVE_MAX_MS; + private volatile long announcedTimeoutMs = -1; + private volatile int openRequests; + private volatile String lastError; + private volatile String videoParam; + private volatile String requested; + private final ArrayDeque lastNotifies = new ArrayDeque<>(); + private final Map topics = new LinkedHashMap<>(); + private final Reader reader = new Reader(); + private int ctpLogged; + + // Video socket; written by its thread + private volatile long lastVideoAt; + private volatile int packets; + private volatile long bytes; + private volatile int rtpPackets; + private volatile int rtpFrames; + private volatile int otherRtp; + private volatile int nonRtp; + private volatile int rawJpegs; + private volatile int foreign; + private volatile String firstPacket; + private volatile String lastSource; + private volatile boolean h264Logged; + private final Map payloadTypes = new LinkedHashMap<>(); + private final Map sources = new LinkedHashMap<>(); + private final Map diagnostic = new LinkedHashMap<>(); + private final Map diagnosticFirst = new LinkedHashMap<>(); + private int logged; + + Client(String tag, Url url, TcpSetup tcpSetup, FlowUfo.SocketSetup udpSetup) { + this.tag = tag; + this.url = url; + this.tcpSetup = tcpSetup; + this.udpSetup = udpSetup; + } + + /** + * Ends run() from any thread. The video sockets close at once; a connected CTP socket + * is left to the session thread, which sends CLOSE_RT_STREAM first (within TICK_MS). + */ + @Override + public void close() { + closed = true; + synchronized (lock) { + for (DatagramSocket s : udpSockets) s.close(); + if (tcp != null && !tcpConnected) { + try { + tcp.close(); + } catch (IOException ignored) {} + } + } + } + + private String withHint(String text) { + String h = hint; + return h != null ? text + ". " + h : text; + } + + /** Milliseconds since the last video packet from the toy, or Long.MAX_VALUE. */ + long videoAgeMs() { + long last = lastVideoAt; + return last > 0 ? now() - last : Long.MAX_VALUE; + } + + String connectingText() { + return withHint("Connecting to the toy (CTP " + url.endpoint() + ")"); + } + + String requestedText() { + return withHint("Stream requested, waiting for video on UDP " + url.rtpPort); + } + + String noVideoText() { + return withHint("No video on UDP " + url.rtpPort + " yet: the toy accepted the" + + " request; see /log"); + } + + String quietText() { + return withHint("The toy's video stopped: waiting on UDP " + url.rtpPort); + } + + void run(UdpReceiver.Condition keepGoing, Listener listener) throws IOException { + InetAddress a = InetAddress.getByName(url.host); + address = a; + hint = subnetHint(a, localForHint != null ? localForHint : FlowUfo.localAddresses()); + if (hint != null) Log.w(TAG, tag + ": Jieli: " + hint); + OutputStream out = null; + boolean requestedStream = false; + Socket s = new Socket(); + try { + // The video port first: nothing the toy sends after OPEN_RT_STREAM is lost + DatagramSocket video = openUdp(url.rtpPort, a); + startThread(video, true, listener); + state = "connecting"; + listener.onState(false, connectingText()); + if (tcpSetup != null) networkTcp = tcpSetup.prepare(s, a); + synchronized (lock) { + tcp = s; + if (closed) return; + } + Log.i(TAG, tag + ": Jieli: connecting to CTP " + url.endpoint() + " (network: " + + networkTcp + "), video on UDP " + url.rtpPort + " (network: " + + networkUdp + ")"); + try { + s.connect(new InetSocketAddress(a, url.port), CONNECT_TIMEOUT_MS); + } catch (IOException e) { + if (closed) return; + throw new IOException("Cannot reach the toy's CTP port " + url.endpoint() + + ": " + e.getMessage()); + } + synchronized (lock) { + tcpConnected = true; + if (closed) return; + } + s.setTcpNoDelay(true); + s.setSoTimeout(TICK_MS); + out = s.getOutputStream(); + InputStream in = s.getInputStream(); + state = "ctp"; + Log.i(TAG, tag + ": Jieli: CTP connected to " + url.endpoint()); + send(out, APP_ACCESS, put("ver", "25", "type", "0")); + long t0 = now(); + long lastRx = t0; + // From the last one sent: an interval announced later applies at once (Qodo + // on #198) + long lastKeepAlive = 0; + long openAt = 0; + boolean diagnostics = false; + boolean receiving = false; + int reopens = 0; + byte[] buf = new byte[8192]; + while (!closed && keepGoing.holds()) { + long t = now(); + if (!requestedStream && (videoParam != null || t - t0 >= paramWaitMs)) { + send(out, DATE_TIME, put("date", new SimpleDateFormat("yyyyMMddHHmmss", + Locale.ROOT).format(new Date()))); + keepAlive(out); + lastKeepAlive = t; + requestStream(out); + requestedStream = true; + openAt = t; + state = "requested"; + listener.onState(false, requestedText()); + } + if (requestedStream && t - lastKeepAlive >= keepAliveIntervalMs) { + keepAlive(out); + lastKeepAlive = t; + } + if (requestedStream) { + long last = lastVideoAt; + boolean flowing = last > 0 && t - last < quietMs; + if (flowing && !receiving) { + receiving = true; + state = "playing"; + Log.i(TAG, tag + ": Jieli: video arrives on UDP " + url.rtpPort + + " from " + lastSource); + listener.onState(true, "Receiving from the toy on UDP " + + url.rtpPort + "…"); + } else if (!flowing && receiving) { + receiving = false; + state = "quiet"; + Log.w(TAG, tag + ": Jieli: no video for " + quietMs + " ms (" + + summary() + ")"); + listener.onState(false, quietText()); + } + if (!diagnostics && last == 0 && t - openAt >= noVideoMs) { + diagnostics = true; + Log.w(TAG, tag + ": Jieli: no video on UDP " + url.rtpPort + " " + + noVideoMs + " ms after OPEN_RT_STREAM (" + summary() + + "); listening on " + ports(diagnosticPorts) + + " too, for the log only"); + for (int p : diagnosticPorts) { + if (p == url.rtpPort) continue; + try { + startThread(openUdp(p, a), false, listener); + } catch (IOException e) { + Log.w(TAG, tag + ": Jieli: cannot listen on UDP " + p + ": " + + e.getMessage()); + } + } + listener.onState(false, noVideoText()); + } + if (!receiving && t - Math.max(openAt, last) >= reopenMs) { + reopens++; + if (reopens <= 3 || reopens % 10 == 0) { + Log.w(TAG, tag + ": Jieli: still no video, OPEN_RT_STREAM again (" + + reopens + ")"); + } + requestStream(out); + openAt = t; + } + } + if (t - lastRx > Math.max(ctpSilenceMs, announcedTimeoutMs)) { + throw new IOException("The toy stopped answering on CTP " + + url.endpoint()); + } + int n; + try { + n = in.read(buf); + } catch (SocketTimeoutException e) { + continue; + } + if (n < 0) throw new IOException("The toy closed the CTP connection"); + lastRx = now(); + reader.feed(buf, 0, n); + Message m; + while ((m = reader.next()) != null) handle(m); + } + } catch (IOException e) { + lastError = e.getMessage(); + if (closed) return; + throw e; + } finally { + if (out != null && requestedStream && !s.isClosed()) { + try { + send(out, CLOSE_RT_STREAM, put("status", "1")); + } catch (IOException ignored) {} + } + try { + s.close(); + } catch (IOException ignored) {} + synchronized (lock) { + for (DatagramSocket d : udpSockets) d.close(); + udpSockets.clear(); + tcp = null; + tcpConnected = false; + } + Log.i(TAG, tag + ": Jieli: session ended (" + summary() + ")"); + } + } + + private static String ports(int[] p) { + StringBuilder sb = new StringBuilder("UDP"); + for (int i = 0; i < p.length; i++) sb.append(i == 0 ? " " : ", ").append(p[i]); + return sb.toString(); + } + + private DatagramSocket openUdp(int port, InetAddress target) throws IOException { + DatagramSocket d = new DatagramSocket(null); + try { + d.bind(new InetSocketAddress(port)); + } catch (IOException e) { + d.close(); + throw new IOException("UDP port " + port + " is in use (another jieli:// slot" + + " or app?): " + e.getMessage()); + } + try { + d.setReceiveBufferSize(1024 * 1024); + } catch (IOException ignored) {} + String n = udpSetup != null ? udpSetup.prepare(d, target) : "default"; + if (port == url.rtpPort) networkUdp = n; + synchronized (lock) { + udpSockets.add(d); + if (closed) d.close(); + } + return d; + } + + private void startThread(DatagramSocket d, boolean video, Listener listener) { + int port = d.getLocalPort(); + Thread th = new Thread(() -> { + byte[] b = new byte[65536]; + DatagramPacket p = new DatagramPacket(b, b.length); + try { + while (!closed && !d.isClosed()) { + p.setLength(b.length); + d.receive(p); + if (video) { + video(p, listener); + } else { + diagnostic(port, p); + } + } + } catch (IOException e) { + // Closed + } + }, tag + "-jieli-" + port); + th.setDaemon(true); + th.start(); + } + + private void send(OutputStream out, String topic, String json) throws IOException { + out.write(encode(topic, json)); + out.flush(); + ctpSent++; + if (!KEEP_ALIVE.equals(topic) || keepAlives < 2) logCtp("→", topic, json); + } + + private void keepAlive(OutputStream out) throws IOException { + send(out, KEEP_ALIVE, put()); + keepAlives++; + } + + private void requestStream(OutputStream out) throws IOException { + int[] p = videoParams(); + int w = url.width > 0 ? url.width : p[0] > 0 ? p[0] : DEFAULT_WIDTH; + int h = url.height > 0 ? url.height : p[1] > 0 ? p[1] : DEFAULT_HEIGHT; + int fps = url.fps > 0 ? url.fps : p[2] > 0 ? p[2] : DEFAULT_FPS; + requested = w + "x" + h + " at " + fps + " fps, format " + url.format + + (url.format == FORMAT_H264 ? " (H.264)" : " (JPEG)"); + send(out, OPEN_RT_STREAM, openStream(url.format, w, h, fps)); + openRequests++; + } + + /** Width, height and fps of VIDEO_PARAM, -1 each when not received. */ + private int[] videoParams() { + Map j = parseJson(videoParamJson); + return new int[] {intParam(j, "w"), intParam(j, "h"), intParam(j, "fps")}; + } + + private volatile String videoParamJson; + + private void logCtp(String dir, String topic, String json) { + int n = ++ctpLogged; + if (n <= 60 || n % 100 == 0) { + Log.i(TAG, tag + ": Jieli: CTP " + dir + " " + topic + + (json.isEmpty() ? "" : " " + truncate(json, 200))); + } + } + + private void handle(Message m) { + ctpReceived++; + synchronized (topics) { + Integer c = topics.get(m.topic); + topics.put(m.topic, c == null ? 1 : c + 1); + if (!KEEP_ALIVE.equals(m.topic)) { + lastNotifies.addLast(truncate(m.toString(), 200)); + while (lastNotifies.size() > 8) lastNotifies.removeFirst(); + } + } + if (KEEP_ALIVE.equals(m.topic)) { + keepAliveReplies++; + if (keepAliveReplies <= 2) logCtp("←", m.topic, m.json); + return; + } + logCtp("←", m.topic, m.json); + Map j = parseJson(m.json); + if (KEEP_ALIVE_INTERVAL.equals(m.topic)) { + int timeout = intParam(j, "timeout"); + if (timeout > 0) { + announcedTimeoutMs = timeout; + keepAliveIntervalMs = keepAliveMs(timeout); + Log.i(TAG, tag + ": Jieli: keepalive every " + keepAliveIntervalMs + + " ms (timeout " + timeout + " ms)"); + } + } else if (VIDEO_PARAM.equals(m.topic)) { + int w = intParam(j, "w"); + int h = intParam(j, "h"); + if (w > 0 && h > 0) { + videoParamJson = m.json; + videoParam = w + "x" + h + " at " + intParam(j, "fps") + " fps, format " + + intParam(j, "format"); + } + } + } + + private void video(DatagramPacket p, Listener listener) { + byte[] b = p.getData(); + int len = p.getLength(); + packets++; + bytes += len; + String from = p.getAddress().getHostAddress() + ":" + p.getPort(); + lastSource = from; + synchronized (sources) { + Integer c = sources.get(from); + if (c != null || sources.size() < 8) { + sources.put(from, c == null ? 1 : c + 1); + } else { + Integer o = sources.get("other"); + sources.put("other", o == null ? 1 : o + 1); + } + } + boolean rtp = len >= 12 && (b[0] & 0xC0) == 0x80; + int pt = rtp ? b[1] & 0x7F : -1; + boolean toy = p.getAddress().equals(address); + if (firstPacket == null) { + firstPacket = from + ", " + len + " bytes: " + FlowUfo.hex(b, 0, Math.min(len, 32)); + } + int n = ++logged; + if (n <= 8 || n % 1000 == 0) { + Log.i(TAG, tag + ": Jieli: UDP " + url.rtpPort + " packet " + n + " from " + from + + (toy ? "" : " (not the toy: ignored)") + ", " + len + " bytes, " + + (rtp ? "RTP PT " + pt + " seq " + ((b[2] & 0xFF) << 8 | (b[3] & 0xFF)) + + ((b[1] & 0x80) != 0 ? " marker" : "") : "not RTP") + + ": " + FlowUfo.hex(b, 0, Math.min(len, 32))); + } + if (!toy) { + foreign++; + return; + } + if (rtp) { + synchronized (payloadTypes) { + Integer c = payloadTypes.get(pt); + payloadTypes.put(pt, c == null ? 1 : c + 1); + } + } + if (rtp && (pt == PT_JPEG || pt == PT_H264)) { + rtpPackets++; + if ((b[1] & 0x80) != 0) rtpFrames++; + if (pt == PT_H264 && !h264Logged) { + h264Logged = true; + Log.i(TAG, tag + ": Jieli: RTP payload type 96: played as H.264"); + } + lastVideoAt = now(); + listener.onRtp(b, 0, len); + } else if (!rtp && len >= 4 && (b[0] & 0xFF) == 0xFF && (b[1] & 0xFF) == 0xD8 + && (b[len - 2] & 0xFF) == 0xFF && (b[len - 1] & 0xFF) == 0xD9) { + // A whole JPEG without RTP: turbodrone takes those too + rawJpegs++; + lastVideoAt = now(); + listener.onJpeg(b, 0, len); + } else if (rtp) { + if (++otherRtp <= 3) { + Log.w(TAG, tag + ": Jieli: RTP payload type " + pt + " on UDP " + + url.rtpPort + ": not video we know, ignored"); + } + } else { + nonRtp++; + } + } + + private void diagnostic(int port, DatagramPacket p) { + int count; + synchronized (diagnostic) { + Integer c = diagnostic.get(port); + count = c == null ? 1 : c + 1; + diagnostic.put(port, count); + if (count == 1) { + diagnosticFirst.put(port, p.getAddress().getHostAddress() + ":" + p.getPort() + + ", " + p.getLength() + " bytes: " + + FlowUfo.hex(p.getData(), 0, Math.min(p.getLength(), 32))); + } + } + if (count <= 5 || count % 1000 == 0) { + byte[] b = p.getData(); + int len = p.getLength(); + boolean rtp = len >= 12 && (b[0] & 0xC0) == 0x80; + Log.i(TAG, tag + ": Jieli: diagnostic UDP " + port + " packet " + count + " from " + + p.getAddress().getHostAddress() + ":" + p.getPort() + ", " + len + + " bytes" + (rtp ? ", RTP PT " + (b[1] & 0x7F) : "") + ": " + + FlowUfo.hex(b, 0, Math.min(len, 32)) + " (not played)"); + } + } + + private String summary() { + return String.format(Locale.ROOT, "CTP %d sent, %d received, %d keepalives;" + + " UDP %d: %d packets, %d RTP video, %d frames, %d not RTP," + + " %d from others", ctpSent, ctpReceived, keepAlives, url.rtpPort, + packets, rtpPackets, rtpFrames, nonRtp, foreign); + } + + /** For /stats.json. */ + Map stats() { + Map m = new LinkedHashMap<>(); + m.put("state", state); + m.put("target", url.endpoint()); + m.put("network", networkTcp); + if (hint != null) m.put("hint", hint); + Map ctp = new LinkedHashMap<>(); + ctp.put("sent", ctpSent); + ctp.put("received", ctpReceived); + ctp.put("keepalives", keepAlives); + ctp.put("keepalive_replies", keepAliveReplies); + ctp.put("keepalive_ms", keepAliveIntervalMs); + if (announcedTimeoutMs > 0) ctp.put("announced_timeout_ms", announcedTimeoutMs); + ctp.put("open_requests", openRequests); + if (reader.skipped > 0) ctp.put("skipped_bytes", reader.skipped); + synchronized (topics) { + ctp.put("topics", new LinkedHashMap<>(topics)); + ctp.put("last_notifies", new ArrayList<>(lastNotifies)); + } + m.put("ctp", ctp); + if (videoParam != null) m.put("video_param", videoParam); + if (requested != null) m.put("requested", requested); + Map rtp = new LinkedHashMap<>(); + rtp.put("port", url.rtpPort); + rtp.put("network", networkUdp); + rtp.put("packets", packets); + rtp.put("bytes", bytes); + rtp.put("video_packets", rtpPackets); + rtp.put("frames", rtpFrames); + if (rawJpegs > 0) rtp.put("raw_jpegs", rawJpegs); + rtp.put("not_rtp", nonRtp); + if (otherRtp > 0) rtp.put("other_rtp", otherRtp); + if (foreign > 0) rtp.put("from_others", foreign); + synchronized (payloadTypes) { + Map pts = new LinkedHashMap<>(); + for (Map.Entry e : payloadTypes.entrySet()) { + pts.put(String.valueOf(e.getKey()), e.getValue()); + } + rtp.put("payload_types", pts); + } + if (firstPacket != null) rtp.put("first_packet", firstPacket); + m.put("rtp", rtp); + synchronized (sources) { + m.put("sources", new LinkedHashMap<>(sources)); + } + synchronized (diagnostic) { + if (!diagnostic.isEmpty()) { + Map d = new LinkedHashMap<>(); + for (Map.Entry e : diagnostic.entrySet()) { + Map one = new LinkedHashMap<>(); + one.put("packets", e.getValue()); + one.put("first", diagnosticFirst.get(e.getKey())); + d.put(String.valueOf(e.getKey()), one); + } + m.put("diagnostic_ports", d); + } + } + if (lastError != null) m.put("last_error", lastError); + return m; + } + } + + private static long now() { + return System.nanoTime() / 1_000_000; + } +} diff --git a/android/app/src/main/java/com/openipc/decoder/NetInfo.java b/android/app/src/main/java/com/openipc/decoder/NetInfo.java index 5c7f0790..95ef3712 100644 --- a/android/app/src/main/java/com/openipc/decoder/NetInfo.java +++ b/android/app/src/main/java/com/openipc/decoder/NetInfo.java @@ -20,6 +20,7 @@ import java.net.Inet6Address; import java.net.InetAddress; import java.net.NetworkInterface; +import java.net.Socket; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -74,27 +75,62 @@ static boolean isLocalClient(InetAddress a) { * Qodo on #188), else the interface bound to. */ static String bindToNetworkOf(DatagramSocket socket, InetAddress target) { - Context ctx = App.context; - if (ctx == null || Build.VERSION.SDK_INT < Build.VERSION_CODES.M) { + if (Build.VERSION.SDK_INT < Build.VERSION_CODES.M) return "default"; + try { + Found f = networkOf(target); + if (f == null) return "default"; + f.network.bindSocket(socket); + return "bound to " + f.name; + } catch (Exception e) { + Log.w(TAG, "No network bound for " + target.getHostAddress() + ": " + e.getMessage()); return "default"; } + } + + /** + * The same for a TCP socket that is not connected yet (the CTP connection of a jieli:// + * slot, #187); binding one needs Android 6. + */ + static String bindToNetworkOf(Socket socket, InetAddress target) { + if (Build.VERSION.SDK_INT < Build.VERSION_CODES.M) return "default"; try { - ConnectivityManager cm = - (ConnectivityManager) ctx.getSystemService(Context.CONNECTIVITY_SERVICE); - if (cm == null) return "default"; - for (Network n : cm.getAllNetworks()) { - LinkProperties lp = cm.getLinkProperties(n); - if (lp == null) continue; - for (LinkAddress la : lp.getLinkAddresses()) { - if (!sameSubnet(la.getAddress(), la.getPrefixLength(), target)) continue; - n.bindSocket(socket); - return "bound to " + lp.getInterfaceName(); - } - } + Found f = networkOf(target); + if (f == null) return "default"; + f.network.bindSocket(socket); + return "bound to " + f.name; } catch (Exception e) { Log.w(TAG, "No network bound for " + target.getHostAddress() + ": " + e.getMessage()); + return "default"; + } + } + + private static final class Found { + final Network network; + final String name; + + Found(Network network, String name) { + this.network = network; + this.name = name; + } + } + + /** The network whose own subnet holds {@code target}, or null. */ + private static Found networkOf(InetAddress target) { + Context ctx = App.context; + if (ctx == null) return null; + ConnectivityManager cm = + (ConnectivityManager) ctx.getSystemService(Context.CONNECTIVITY_SERVICE); + if (cm == null) return null; + for (Network n : cm.getAllNetworks()) { + LinkProperties lp = cm.getLinkProperties(n); + if (lp == null) continue; + for (LinkAddress la : lp.getLinkAddresses()) { + if (sameSubnet(la.getAddress(), la.getPrefixLength(), target)) { + return new Found(n, lp.getInterfaceName()); + } + } } - return "default"; + return null; } /** {@code a} and {@code b} share the first {@code prefix} bits (and the address family). */ diff --git a/android/app/src/main/java/com/openipc/decoder/QuadCell.java b/android/app/src/main/java/com/openipc/decoder/QuadCell.java index d9e71e8b..d1c49a2f 100644 --- a/android/app/src/main/java/com/openipc/decoder/QuadCell.java +++ b/android/app/src/main/java/com/openipc/decoder/QuadCell.java @@ -431,6 +431,9 @@ static boolean isKeyFrameNal(int type, boolean h265) { /** FLOW-UFO statistics of a flowufo:// cell (#187), else null. */ java.util.Map flowUfoStats() { return session.flowUfoStats(); } + /** Jieli statistics of a jieli:// cell (#187), else null. */ + java.util.Map jieliStats() { return session.jieliStats(); } + /** The cooingdv heartbeat of an RTSP cell once it used one (#187), else null. */ java.util.Map heartbeatStats() { return session.heartbeatStats(); } diff --git a/android/app/src/main/java/com/openipc/decoder/RtspClient.java b/android/app/src/main/java/com/openipc/decoder/RtspClient.java index c40e424f..940a6010 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspClient.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspClient.java @@ -89,6 +89,8 @@ void configure(String host, String userAgent, Transport.Mode transport) { long getBitrateMarkNs() { return session.getBitrateMarkNs(); } /** FLOW-UFO statistics while a flowufo:// slot plays (#187), else null. */ java.util.Map flowUfoStats() { return session.flowUfoStats(); } + /** Jieli statistics while a jieli:// slot plays (#187), else null. */ + java.util.Map jieliStats() { return session.jieliStats(); } /** The cooingdv heartbeat once this slot used one (#187), else null. */ java.util.Map heartbeatStats() { return session.heartbeatStats(); } diff --git a/android/app/src/main/java/com/openipc/decoder/RtspSession.java b/android/app/src/main/java/com/openipc/decoder/RtspSession.java index bad7a2c9..fa7c55e6 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -24,6 +24,7 @@ import java.util.Locale; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; /** @@ -107,6 +108,8 @@ default void onDetail(String text) {} private volatile boolean listening; /** Client of a flowufo:// slot (#187), else null: it probes on its own when silent. */ private volatile FlowUfo.Client ufo; + /** Client of a jieli:// slot (#187), else null: it asks for the stream again on its own. */ + private volatile Jieli.Client jieli; /** Heartbeat of the RTSP connection now (#187), else null; counts of the earlier ones. */ private volatile UdpHeartbeat heartbeat; private volatile int heartbeatsBefore; @@ -216,6 +219,7 @@ void stop() { private void closeSocketLocked() { if (listener != null) listener.close(); if (ufo != null) ufo.close(); + if (jieli != null) jieli.close(); if (heartbeat != null) heartbeat.close(); if (udp != null) { udp.close(); @@ -289,7 +293,7 @@ private void watchdogLoop(int gen) { } } } - if (active && !listening && ufo == null + if (active && !listening && ufo == null && jieli == null && System.nanoTime() / 1_000_000 - lastPacketMs > watchdogMs) { Log.w(TAG, tag + ": stream is inactive, reconnecting"); active = false; @@ -310,6 +314,7 @@ private boolean connect(int gen) throws IOException { if (RtpListener.Url.matches(url)) return listen(gen, RtpListener.Url.parse(url)); if (HttpMjpeg.matches(url)) return httpMjpeg(gen, HttpMjpeg.Url.parse(url)); if (FlowUfo.Url.matches(url)) return flowUfo(gen, FlowUfo.Url.parse(url)); + if (Jieli.Url.matches(url)) return jieli(gen, Jieli.Url.parse(url)); Url u = Url.parse(url); Socket s = new Socket(); synchronized (lock) { @@ -724,6 +729,111 @@ public void onFrame(byte[] buf, int length) { } } + /** Jieli statistics for /stats.json while a jieli:// slot runs (#187), else null. */ + Map jieliStats() { + Jieli.Client c = jieli; + return c != null ? c.stats() : null; + } + + /** + * A jieli:// slot (#187): CTP over TCP asks the toy for the stream, which comes as RTP to + * a fixed local UDP port: payload type 26 goes to the RTP/JPEG path, 96 to the H.264 one, + * as on a udp:// slot. The client asks again when no video comes, so the watchdog leaves + * it alone. Returns true once it ran, like a PLAY that succeeded. + */ + private boolean jieli(int gen, Jieli.Url u) throws IOException { + Jieli.Client c = new Jieli.Client(tag, u, NetInfo::bindToNetworkOf, + NetInfo::bindToNetworkOf); + synchronized (lock) { + if (gen != generation) return false; + jieli = c; + keepalive = null; + lastPacketMs = System.nanoTime() / 1_000_000; + streamBytes = 0; + bitrateMarkNs = System.nanoTime(); + active = true; + } + try { + final int[] pt = {u.format == Jieli.FORMAT_H264 ? Jieli.PT_H264 : Jieli.PT_JPEG}; + callback.onSdp(jieliSdp(pt[0])); + // onState comes from the CTP thread, the video from the UDP one + final AtomicBoolean first = new AtomicBoolean(true); + final AtomicBoolean received = new AtomicBoolean(false); + final Object order = new Object(); + c.run(() -> active && gen == generation, new Jieli.Listener() { + @Override + public void onState(boolean receiving, String text) { + if (gen != generation) return; + // In order with usable() on the UDP thread: a packet that came in since + // the CTP thread saw the video stop wins (Qodo on #198) + synchronized (order) { + if (receiving) { + inUse = "Jieli RTP"; + received.set(true); + } else { + if (c.videoAgeMs() < Jieli.quietMs) return; + inUse = null; + // Video that stopped says so; the next frame reports "connected" + status(gen, received.get() ? "disconnected" : "buffering"); + first.set(true); + } + callback.onDetail(text); + } + } + + @Override + public void onRtp(byte[] buf, int offset, int length) { + if (gen != generation) return; + int type = buf[offset + 1] & 0x7F; + if (type != pt[0]) { + // The toy sends the other codec than asked for: follow it + pt[0] = type; + Log.i(TAG, tag + ": Jieli: video is " + + (type == Jieli.PT_H264 ? "H.264" : "JPEG") + " (RTP PT " + type + + ")"); + callback.onSdp(jieliSdp(type)); + } + streamBytes += length; + if (callback.onPacket(CHANNEL_VIDEO, buf, offset, length)) usable(); + } + + @Override + public void onJpeg(byte[] buf, int offset, int length) { + if (gen != generation) return; + if (pt[0] != Jieli.PT_JPEG) { + pt[0] = Jieli.PT_JPEG; + callback.onSdp(jieliSdp(Jieli.PT_JPEG)); + } + streamBytes += length; + if (callback.onJpegFrame(buf, offset, length)) usable(); + } + + private void usable() { + lastPacketMs = System.nanoTime() / 1_000_000; + synchronized (order) { + if (first.compareAndSet(true, false)) status(gen, "connected"); + } + } + }); + return true; + } finally { + synchronized (lock) { + if (jieli == c) { + jieli = null; + inUse = null; + } + } + c.close(); + } + } + + private static Sdp jieliSdp(int pt) { + Sdp sdp = new Sdp(); + sdp.videoPt = pt; + sdp.videoJpeg = pt == Jieli.PT_JPEG; + return sdp; + } + private boolean listen(int gen, RtpListener.Url u) throws IOException { RtpListener l = new RtpListener(tag, u); synchronized (lock) { diff --git a/android/app/src/main/java/com/openipc/decoder/StreamProbes.java b/android/app/src/main/java/com/openipc/decoder/StreamProbes.java index 4c1ddfd6..7a3c08a6 100644 --- a/android/app/src/main/java/com/openipc/decoder/StreamProbes.java +++ b/android/app/src/main/java/com/openipc/decoder/StreamProbes.java @@ -50,7 +50,9 @@ static String skipReason(String url, String defaultUrl, String screenUrl) { // A listening slot would take the port; the camera pushes to it, nothing to ask if (RtpListener.Url.matches(url)) return "UDP listen slot: not checked in single view"; // A toy drone sends to one client; a check would take the stream from the screen - if (FlowUfo.Url.matches(url)) return "toy drone slot: not checked in single view"; + if (FlowUfo.Url.matches(url) || Jieli.Url.matches(url)) { + return "toy drone slot: not checked in single view"; + } String e = endpoint(url); if (e == null) return "invalid URL"; if (e.equals(endpoint(screenUrl))) return "same camera as on screen"; diff --git a/android/app/src/test/java/com/openipc/decoder/FlowUfoTest.java b/android/app/src/test/java/com/openipc/decoder/FlowUfoTest.java index 25d26fd5..a281bff4 100644 --- a/android/app/src/test/java/com/openipc/decoder/FlowUfoTest.java +++ b/android/app/src/test/java/com/openipc/decoder/FlowUfoTest.java @@ -1032,7 +1032,8 @@ public void subnetHints() throws Exception { FlowUfo.subnetHint(ip("192.168.169.2"), on169)); assertEquals("This phone is on 192.168.169.x: the toy is usually flowufo://192.168.169.1", FlowUfo.subnetHint(ip("192.168.1.1"), on169)); - assertEquals("Toys on 192.168.80.x (HF_ Wi-Fi) use another protocol, not supported yet", + assertEquals("This phone is on 192.168.80.x (HF_ Wi-Fi), a Jieli toy: try" + + " jieli://192.168.80.1", FlowUfo.subnetHint(ip("192.168.169.1"), Collections.singletonList(ip("192.168.80.3")))); // On both networks: no guess diff --git a/android/app/src/test/java/com/openipc/decoder/JieliTest.java b/android/app/src/test/java/com/openipc/decoder/JieliTest.java new file mode 100644 index 00000000..0474451a --- /dev/null +++ b/android/app/src/test/java/com/openipc/decoder/JieliTest.java @@ -0,0 +1,614 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * JieliTest.java — jieli:// slots: CTP framing, the session against a fake toy, RTP/JPEG on UDP + * + */ + +package com.openipc.decoder; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +public class JieliTest { + private static final InetAddress LOOPBACK = InetAddress.getLoopbackAddress(); + + @Before + public void shortTimers() { + Jieli.paramWaitMs = 300; + Jieli.diagnosticPorts = new int[0]; + } + + @After + public void restoreTimers() { + Jieli.paramWaitMs = 1500; + Jieli.noVideoMs = 5000; + Jieli.reopenMs = 10_000; + Jieli.quietMs = 3000; + Jieli.ctpSilenceMs = 20_000; + Jieli.diagnosticPorts = new int[] {1234, 2224}; + } + + // ====================== Helpers ====================== + + private static byte[] bytes(String hex) { + String h = hex.replaceAll("\\s", ""); + byte[] b = new byte[h.length() / 2]; + for (int i = 0; i < b.length; i++) { + b[i] = (byte) Integer.parseInt(h.substring(2 * i, 2 * i + 2), 16); + } + return b; + } + + private static byte[] concat(byte[]... parts) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + for (byte[] p : parts) out.write(p, 0, p.length); + return out.toByteArray(); + } + + /** RTP/JPEG packets of a 64x32 test picture (see JpegAssemblerTest). */ + private static List jpegPackets() throws IOException { + ByteArrayOutputStream all = new ByteArrayOutputStream(); + try (InputStream in = JieliTest.class.getResourceAsStream("/rtp/jpeg_q50_64x32.rtp")) { + assertNotNull(in); + byte[] b = new byte[4096]; + int n; + while ((n = in.read(b)) > 0) all.write(b, 0, n); + } + byte[] d = all.toByteArray(); + List out = new ArrayList<>(); + for (int i = 0; i < d.length; ) { + int len = ((d[i] & 0xFF) << 8) | (d[i + 1] & 0xFF); + out.add(Arrays.copyOfRange(d, i + 2, i + 2 + len)); + i += 2 + len; + } + return out; + } + + private static byte[] resource(String name) throws IOException { + ByteArrayOutputStream all = new ByteArrayOutputStream(); + try (InputStream in = JieliTest.class.getResourceAsStream("/rtp/" + name)) { + assertNotNull(in); + byte[] b = new byte[4096]; + int n; + while ((n = in.read(b)) > 0) all.write(b, 0, n); + } + return all.toByteArray(); + } + + private static int freeUdpPort() throws IOException { + try (DatagramSocket s = new DatagramSocket(0)) { + return s.getLocalPort(); + } + } + + private static long now() { + return System.nanoTime() / 1_000_000; + } + + // ====================== Framing ====================== + + @Test + public void encodesAsTheVendorApp() { + // CTP_KEEP_ALIVE {"op":"PUT"} as in the capture of #196 + assertArrayEquals(concat("CTP:".getBytes(StandardCharsets.US_ASCII), bytes("0E 00"), + "CTP_KEEP_ALIVE".getBytes(StandardCharsets.US_ASCII), bytes("0C 00 00 00"), + "{\"op\":\"PUT\"}".getBytes(StandardCharsets.US_ASCII)), + Jieli.encode(Jieli.KEEP_ALIVE, Jieli.put())); + // An empty JSON has length 0 + assertArrayEquals(concat("CTP:".getBytes(StandardCharsets.US_ASCII), bytes("02 00"), + "AB".getBytes(StandardCharsets.US_ASCII), bytes("00 00 00 00")), + Jieli.encode("AB", "")); + } + + @Test + public void commandsAsTheVendorApp() { + assertEquals("{\"op\":\"PUT\",\"param\":{\"ver\":\"25\",\"type\":\"0\"}}", + Jieli.put("ver", "25", "type", "0")); + assertEquals("{\"op\":\"PUT\",\"param\":{\"format\":\"0\",\"h\":\"480\",\"w\":\"640\"," + + "\"fps\":\"25\"}}", Jieli.openStream(0, 640, 480, 25)); + assertEquals("{\"op\":\"PUT\",\"param\":{\"status\":\"1\"}}", Jieli.put("status", "1")); + assertEquals("{\"op\":\"PUT\",\"param\":{\"a\\\"b\":\"c\\\\\"}}", Jieli.put("a\"b", "c\\")); + } + + @Test + public void readsMessagesInPiecesAndTogether() { + byte[] a = Jieli.encode("VIDEO_PARAM", + "{\"op\":\"NOTIFY\",\"param\":{\"w\":\"640\",\"h\":\"480\"}}"); + byte[] b = Jieli.encode(Jieli.KEEP_ALIVE, ""); + byte[] c = Jieli.encode("KEEP_ALIVE_INTERVAL", "{\"param\":{\"timeout\":\"60000\"}}"); + byte[] all = concat(a, b, c); + // Byte by byte + Jieli.Reader r = new Jieli.Reader(); + List got = new ArrayList<>(); + for (int i = 0; i < all.length; i++) { + r.feed(all, i, 1); + Jieli.Message m; + while ((m = r.next()) != null) got.add(m); + } + assertEquals(3, got.size()); + assertEquals("VIDEO_PARAM", got.get(0).topic); + assertEquals("{\"op\":\"NOTIFY\",\"param\":{\"w\":\"640\",\"h\":\"480\"}}", + got.get(0).json); + assertEquals(Jieli.KEEP_ALIVE, got.get(1).topic); + assertEquals("", got.get(1).json); + assertEquals("KEEP_ALIVE_INTERVAL", got.get(2).topic); + assertEquals(0, r.skipped); + // All in one read, split across a length field + r = new Jieli.Reader(); + r.feed(all, 0, a.length + 5); + assertEquals("VIDEO_PARAM", r.next().topic); + assertNull(r.next()); + r.feed(all, a.length + 5, all.length - a.length - 5); + assertEquals(Jieli.KEEP_ALIVE, r.next().topic); + assertEquals("KEEP_ALIVE_INTERVAL", r.next().topic); + assertNull(r.next()); + } + + @Test + public void skipsWhatIsNotCtp() { + byte[] good = Jieli.encode("SD_STATUS", "{\"online\":\"1\"}"); + // Garbage, then a "CTP:" with a topic length out of range, then a real message + byte[] bad = concat("xyCT".getBytes(StandardCharsets.US_ASCII), + "CTP:".getBytes(StandardCharsets.US_ASCII), bytes("FF FF"), good); + Jieli.Reader r = new Jieli.Reader(); + r.feed(bad, 0, bad.length); + Jieli.Message m = r.next(); + assertNotNull(m); + assertEquals("SD_STATUS", m.topic); + assertEquals(10, r.skipped); + assertNull(r.next()); + // A large message grows the buffer + StringBuilder big = new StringBuilder("{\"x\":\""); + for (int i = 0; i < 20000; i++) big.append('a'); + big.append("\"}"); + byte[] large = Jieli.encode("BIG", big.toString()); + r.feed(large, 0, large.length); + assertEquals(big.toString(), r.next().json); + } + + @Test + public void readsJson() { + Map j = Jieli.parseJson("{\"errno\":0,\"op\":\"NOTIFY\",\"param\":" + + "{\"w\":\"640\",\"h\":\"480\",\"format\":\"0\",\"fps\":\"25\",\"l\":[1,\"x\"]}}"); + assertNotNull(j); + assertEquals("NOTIFY", j.get("op")); + assertEquals("0", j.get("errno")); + assertEquals("640", Jieli.param(j, "w")); + assertEquals(25, Jieli.intParam(j, "fps")); + assertEquals("NOTIFY", Jieli.param(j, "op")); + assertEquals(-1, Jieli.intParam(j, "missing")); + assertEquals("a\"b\u00e9", Jieli.parseJson("{\"k\":\"a\\\"b\\u00e9\"}").get("k")); + assertNull(Jieli.parseJson("")); + assertNull(Jieli.parseJson("{\"a\":")); + assertNull(Jieli.parseJson("[1]")); + assertNull(Jieli.parseJson("{} x")); + } + + @Test + public void keepAliveInterval() { + assertEquals(7500, Jieli.keepAliveMs(60000)); + assertEquals(7500, Jieli.keepAliveMs(-1)); + assertEquals(2500, Jieli.keepAliveMs(10000)); + assertEquals(250, Jieli.keepAliveMs(100)); + } + + // ====================== URL ====================== + + @Test + public void urls() throws Exception { + Jieli.Url u = Jieli.Url.parse("jieli://"); + assertEquals("192.168.80.1", u.host); + assertEquals(3333, u.port); + assertEquals(6666, u.rtpPort); + assertEquals(0, u.width); + assertEquals(Jieli.FORMAT_JPEG, u.format); + u = Jieli.Url.parse(" JIELI://192.168.80.1:3334/?w=1280&h=720&fps=30&rtp=7000&format=1 "); + assertEquals(3334, u.port); + assertEquals(1280, u.width); + assertEquals(720, u.height); + assertEquals(30, u.fps); + assertEquals(7000, u.rtpPort); + assertEquals(Jieli.FORMAT_H264, u.format); + assertEquals("192.168.80.1:3334", u.endpoint()); + assertTrue(Jieli.Url.matches("jieli://x")); + assertTrue(!Jieli.Url.matches("flowufo://")); + for (String bad : new String[] {"jieli://h:0", "jieli://h:x", "jieli://?w=8", + "jieli://?fps=0", "jieli://?rtp=70000", "jieli://?format=2", "jieli://a b", + "jieli://[::1"}) { + try { + Jieli.Url.parse(bad); + fail(bad); + } catch (IOException expected) { + // refused + } + } + assertEquals("toy drone slot: not checked in single view", + StreamProbes.skipReason("jieli://", "rtsp://default", null)); + } + + @Test + public void subnetHints() throws Exception { + InetAddress toy = InetAddress.getByName("192.168.80.1"); + List on80 = Collections.singletonList(InetAddress.getByName("192.168.80.2")); + assertNull(Jieli.subnetHint(toy, on80)); + assertEquals("192.168.80.2 is this phone; the toy is usually 192.168.80.1", + Jieli.subnetHint(InetAddress.getByName("192.168.80.2"), on80)); + assertEquals("This phone is on 192.168.80.x: the toy is usually jieli://192.168.80.1", + Jieli.subnetHint(InetAddress.getByName("192.168.169.1"), on80)); + assertEquals("This network looks like a FLOW-UFO toy: try flowufo://192.168.169.1", + Jieli.subnetHint(toy, + Collections.singletonList(InetAddress.getByName("192.168.169.3")))); + assertEquals("This network looks like a KY UFO RTSP toy: try" + + " rtsp://192.168.1.1:7070/webcam", + Jieli.subnetHint(toy, + Collections.singletonList(InetAddress.getByName("192.168.1.100")))); + assertNull(Jieli.subnetHint(toy, Collections.emptyList())); + } + + // ====================== A fake toy on loopback ====================== + + /** One message the fake toy received, with the time. */ + private static final class Got { + final String topic; + final String json; + final long at; + + Got(Jieli.Message m) { + topic = m.topic; + json = m.json; + at = now(); + } + } + + /** + * Answers like the toy of #196: KEEP_ALIVE_INTERVAL and VIDEO_PARAM after APP_ACCESS, + * empty keepalive replies, an OPEN_RT_STREAM notify; then the video to {@code rtpPort}. + */ + private static final class FakeToy implements AutoCloseable { + final ServerSocket server; + final List got = new CopyOnWriteArrayList<>(); + final CountDownLatch closeStream = new CountDownLatch(1); + final CountDownLatch opened = new CountDownLatch(1); + final int timeoutMs; + final boolean sendVideoParam; + volatile int rtpPort; + volatile boolean sendVideo = true; + /** KEEP_ALIVE_INTERVAL only after the first keepalive, not after APP_ACCESS. */ + volatile boolean lateInterval; + private boolean intervalSent; + private final Thread thread; + + FakeToy(int timeoutMs, boolean sendVideoParam) throws IOException { + this.timeoutMs = timeoutMs; + this.sendVideoParam = sendVideoParam; + server = new ServerSocket(0, 1, LOOPBACK); + thread = new Thread(this::serve, "fake-jieli"); + thread.setDaemon(true); + thread.start(); + } + + int port() { + return server.getLocalPort(); + } + + private void serve() { + try (Socket s = server.accept()) { + InputStream in = s.getInputStream(); + OutputStream out = s.getOutputStream(); + Jieli.Reader r = new Jieli.Reader(); + byte[] buf = new byte[4096]; + int n; + while ((n = in.read(buf)) > 0) { + r.feed(buf, 0, n); + Jieli.Message m; + while ((m = r.next()) != null) { + got.add(new Got(m)); + switch (m.topic) { + case Jieli.APP_ACCESS: + if (!lateInterval) interval(out); + out.write(Jieli.encode("BAT_STATUS", + "{\"op\":\"NOTIFY\",\"param\":{\"level\":\"4\"}}")); + if (sendVideoParam) { + out.write(Jieli.encode(Jieli.VIDEO_PARAM, + "{\"op\":\"NOTIFY\",\"param\":{\"w\":\"320\"," + + "\"h\":\"240\",\"format\":\"0\"," + + "\"fps\":\"15\"}}")); + } + break; + case Jieli.KEEP_ALIVE: + out.write(Jieli.encode(Jieli.KEEP_ALIVE, "")); + if (lateInterval && !intervalSent) interval(out); + break; + case Jieli.OPEN_RT_STREAM: + out.write(Jieli.encode(Jieli.OPEN_RT_STREAM, + "{\"op\":\"NOTIFY\",\"param\":{\"status\":\"1\"}}")); + if (opened.getCount() > 0 && sendVideo) sendVideo(); + opened.countDown(); + break; + case Jieli.CLOSE_RT_STREAM: + closeStream.countDown(); + break; + default: + break; + } + out.flush(); + } + } + } catch (IOException e) { + // Closed + } + } + + private void interval(OutputStream out) throws IOException { + intervalSent = true; + out.write(Jieli.encode(Jieli.KEEP_ALIVE_INTERVAL, + "{\"op\":\"NOTIFY\",\"param\":{\"timeout\":\"" + timeoutMs + "\"}}")); + } + + private void sendVideo() throws IOException { + try (DatagramSocket u = new DatagramSocket(0, LOOPBACK)) { + InetSocketAddress to = new InetSocketAddress(LOOPBACK, rtpPort); + // Something that is not RTP first: counted and logged, not played + byte[] junk = {1, 2, 3}; + u.send(new DatagramPacket(junk, junk.length, to)); + for (byte[] p : jpegPackets()) u.send(new DatagramPacket(p, p.length, to)); + } + } + + List topics() { + List t = new ArrayList<>(); + for (Got g : got) t.add(g.topic); + return t; + } + + @Override + public void close() throws IOException { + server.close(); + } + } + + /** Runs the client on a thread; frames assembled from the RTP it delivers. */ + private static final class Running { + final Jieli.Client client; + final AtomicBoolean keepGoing = new AtomicBoolean(true); + final List frames = new CopyOnWriteArrayList<>(); + final List states = new CopyOnWriteArrayList<>(); + final CountDownLatch frame = new CountDownLatch(1); + final CountDownLatch ended = new CountDownLatch(1); + volatile Exception failure; + final Thread thread; + + Running(String url) throws IOException { + client = new Jieli.Client("test", Jieli.Url.parse(url), null, null); + client.localForHint = Collections.emptyList(); + JpegAssembler assembler = new JpegAssembler(new FramePool(4)); + thread = new Thread(() -> { + try { + client.run(keepGoing::get, new Jieli.Listener() { + @Override + public void onState(boolean receiving, String text) { + states.add(text); + } + + @Override + public void onRtp(byte[] buf, int offset, int length) { + assembler.assemble(buf, offset, length, f -> { + frames.add(Arrays.copyOf(f.data(), f.length())); + frame.countDown(); + }); + } + }); + } catch (Exception e) { + failure = e; + } finally { + ended.countDown(); + } + }, "jieli-client"); + thread.start(); + } + } + + @Test + public void sessionAsTheVendorApp() throws Exception { + int rtp = freeUdpPort(); + try (FakeToy toy = new FakeToy(2000, true)) { + toy.rtpPort = rtp; + Running r = new Running("jieli://127.0.0.1:" + toy.port() + "?rtp=" + rtp); + assertTrue("no frame", r.frame.await(5, TimeUnit.SECONDS)); + // A few keepalives at 2000 / 4 = 500 ms + Thread.sleep(1700); + r.keepGoing.set(false); + assertTrue("no CLOSE_RT_STREAM", toy.closeStream.await(3, TimeUnit.SECONDS)); + assertTrue(r.ended.await(3, TimeUnit.SECONDS)); + assertNull(r.failure); + + List t = toy.topics(); + assertEquals(Arrays.asList(Jieli.APP_ACCESS, Jieli.DATE_TIME, Jieli.KEEP_ALIVE, + Jieli.OPEN_RT_STREAM), t.subList(0, 4)); + assertEquals(Jieli.CLOSE_RT_STREAM, t.get(t.size() - 1)); + assertEquals("{\"op\":\"PUT\",\"param\":{\"ver\":\"25\",\"type\":\"0\"}}", + toy.got.get(0).json); + assertTrue(toy.got.get(1).json, + toy.got.get(1).json.matches("\\{\"op\":\"PUT\",\"param\":\\{\"date\":" + + "\"\\d{14}\"}}")); + // The size the toy announced in VIDEO_PARAM + assertEquals(Jieli.openStream(0, 320, 240, 15), toy.got.get(3).json); + assertEquals("{\"op\":\"PUT\",\"param\":{\"status\":\"1\"}}", + toy.got.get(t.size() - 1).json); + assertEquals(1, Collections.frequency(t, Jieli.OPEN_RT_STREAM)); + + List keepAlives = new ArrayList<>(); + for (Got g : toy.got) { + if (g.topic.equals(Jieli.KEEP_ALIVE)) keepAlives.add(g.at); + } + assertTrue("keepalives: " + keepAlives.size(), keepAlives.size() >= 3); + for (int i = 1; i < keepAlives.size(); i++) { + long gap = keepAlives.get(i) - keepAlives.get(i - 1); + assertTrue("gap " + gap, gap >= 400 && gap <= 900); + } + + // The JPEG the RTP/JPEG path makes of the packets + assertArrayEquals(resource("jpeg_q50_64x32.jpg"), r.frames.get(0)); + Map st = r.client.stats(); + @SuppressWarnings("unchecked") + Map ctp = (Map) st.get("ctp"); + assertEquals(500L, ctp.get("keepalive_ms")); + assertEquals("320x240 at 15 fps, format 0", st.get("video_param")); + @SuppressWarnings("unchecked") + Map rtpStats = (Map) st.get("rtp"); + assertEquals(1, rtpStats.get("not_rtp")); + assertTrue((Integer) rtpStats.get("frames") >= 1); + @SuppressWarnings("unchecked") + Map sources = (Map) st.get("sources"); + assertEquals(1, sources.size()); + assertTrue(r.states.get(0), r.states.get(0).startsWith("Connecting to the toy (CTP")); + assertTrue(r.states.contains("Stream requested, waiting for video on UDP " + rtp)); + } + } + + @Test + public void defaultSizeAndCloseOnStop() throws Exception { + int rtp = freeUdpPort(); + try (FakeToy toy = new FakeToy(60000, false)) { + toy.rtpPort = rtp; + toy.sendVideo = false; + Running r = new Running("jieli://127.0.0.1:" + toy.port() + "?rtp=" + rtp); + assertTrue(toy.opened.await(5, TimeUnit.SECONDS)); + // No VIDEO_PARAM: 640x480 at 25 after the wait + Got open = null; + for (Got g : toy.got) if (g.topic.equals(Jieli.OPEN_RT_STREAM)) open = g; + assertNotNull(open); + assertEquals(Jieli.openStream(0, 640, 480, 25), open.json); + assertTrue(open.at - toy.got.get(0).at >= 250); + // close() as RtspSession.stop() does it: CLOSE_RT_STREAM still goes out + r.client.close(); + assertTrue("no CLOSE_RT_STREAM", toy.closeStream.await(3, TimeUnit.SECONDS)); + assertTrue(r.ended.await(3, TimeUnit.SECONDS)); + assertNull(r.failure); + @SuppressWarnings("unchecked") + Map ctp = (Map) r.client.stats().get("ctp"); + assertEquals(7500L, ctp.get("keepalive_ms")); + } + } + + @Test + public void lateIntervalAppliesAtOnce() throws Exception { + int rtp = freeUdpPort(); + try (FakeToy toy = new FakeToy(2000, true)) { + toy.rtpPort = rtp; + toy.sendVideo = false; + toy.lateInterval = true; + Running r = new Running("jieli://127.0.0.1:" + toy.port() + "?rtp=" + rtp); + assertTrue(toy.opened.await(5, TimeUnit.SECONDS)); + long until = now() + 3000; + while (now() < until && Collections.frequency(toy.topics(), Jieli.KEEP_ALIVE) < 2) { + Thread.sleep(20); + } + r.keepGoing.set(false); + assertTrue(r.ended.await(3, TimeUnit.SECONDS)); + List keepAlives = new ArrayList<>(); + for (Got g : toy.got) { + if (g.topic.equals(Jieli.KEEP_ALIVE)) keepAlives.add(g.at); + } + // The second one 500 ms after the first, not 7.5 s + assertTrue("keepalives: " + keepAlives.size(), keepAlives.size() >= 2); + long gap = keepAlives.get(1) - keepAlives.get(0); + assertTrue("gap " + gap, gap >= 400 && gap <= 900); + } + } + + @Test + public void noVideoListensElsewhereAndAsksAgain() throws Exception { + int rtp = freeUdpPort(); + int diag = freeUdpPort(); + Jieli.noVideoMs = 300; + Jieli.reopenMs = 700; + Jieli.diagnosticPorts = new int[] {diag}; + try (FakeToy toy = new FakeToy(60000, true)) { + toy.rtpPort = rtp; + toy.sendVideo = false; + Running r = new Running("jieli://127.0.0.1:" + toy.port() + "?rtp=" + rtp); + assertTrue(toy.opened.await(5, TimeUnit.SECONDS)); + long until = now() + 3000; + while (now() < until && !r.states.contains(r.client.noVideoText())) Thread.sleep(20); + assertTrue(r.states.toString(), r.states.contains(r.client.noVideoText())); + Thread.sleep(100); + try (DatagramSocket u = new DatagramSocket(0, LOOPBACK)) { + byte[] p = {(byte) 0x80, 97, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 9}; + u.send(new DatagramPacket(p, p.length, new InetSocketAddress(LOOPBACK, diag))); + } + until = now() + 3000; + while (now() < until && (Collections.frequency(toy.topics(), Jieli.OPEN_RT_STREAM) < 2 + || r.client.stats().get("diagnostic_ports") == null)) { + Thread.sleep(20); + } + assertTrue(Collections.frequency(toy.topics(), Jieli.OPEN_RT_STREAM) >= 2); + Map st = r.client.stats(); + @SuppressWarnings("unchecked") + Map d = (Map) st.get("diagnostic_ports"); + assertNotNull(d); + @SuppressWarnings("unchecked") + Map one = (Map) d.get(String.valueOf(diag)); + assertEquals(1, one.get("packets")); + r.keepGoing.set(false); + assertTrue(r.ended.await(3, TimeUnit.SECONDS)); + assertNull(r.failure); + // Nothing from a diagnostic port reaches the video path + assertTrue(r.frames.isEmpty()); + } + } + + @Test + public void unreachableToyFails() throws Exception { + int port; + try (ServerSocket s = new ServerSocket(0, 1, LOOPBACK)) { + port = s.getLocalPort(); + } + Jieli.Client c = new Jieli.Client("test", + Jieli.Url.parse("jieli://127.0.0.1:" + port + "?rtp=" + freeUdpPort()), null, null); + c.localForHint = Collections.emptyList(); + try { + c.run(() -> true, new Jieli.Listener() { + @Override + public void onState(boolean receiving, String text) {} + + @Override + public void onRtp(byte[] buf, int offset, int length) {} + }); + fail("connected"); + } catch (IOException e) { + assertTrue(e.getMessage(), + e.getMessage().startsWith("Cannot reach the toy's CTP port")); + } + } +} diff --git a/docs/debug.md b/docs/debug.md index 0ea44458..27fb7b78 100644 --- a/docs/debug.md +++ b/docs/debug.md @@ -168,7 +168,7 @@ Single view and U view: every cell of the grid as the debug page shows it (3×3 | `codec`, `size` | when found | | `fps`, `kbps`, `key_frames` | measured during the check | -The FLOW-UFO test build ([toys.md](toys.md#flow-ufo)) adds a `flowufo` object per cell and in `single`, and a `heartbeat` object for RTSP slots that send the KY UFO heartbeat ([toys.md](toys.md#ky-ufo-rtsp-toys)); it is not in the releases. +The FLOW-UFO test build ([toys.md](toys.md#flow-ufo)) adds a `flowufo` object per cell and in `single` (`jieli` for `jieli://` slots, [toys.md](toys.md#hf--jieli-toys)), and a `heartbeat` object for RTSP slots that send the KY UFO heartbeat ([toys.md](toys.md#ky-ufo-rtsp-toys)); it is not in the releases. ## Log diff --git a/docs/toys.md b/docs/toys.md index ace5f21c..8ea89c50 100644 --- a/docs/toys.md +++ b/docs/toys.md @@ -5,6 +5,7 @@ - [Toy families by Wi‑Fi name](#toy-families-by-wifi-name) - [FLOW-UFO](#flow-ufo) - [KY UFO RTSP toys](#ky-ufo-rtsp-toys) +- [HF / Jieli toys](#hf--jieli-toys) - [Credits](#credits) - [An unknown toy](#an-unknown-toy) - [What to send in a report](#what-to-send-in-a-report) @@ -24,7 +25,7 @@ Cheap drones, cars and other toys with a Wi‑Fi camera usually work the same wa - The video is often **not a standard stream**: no RTSP, no plain RTP. Many toys send JPEG frames over UDP in their own framing and start only after the client has sent a request; some also expect regular replies while they stream. A standard player then sees nothing, and this app stays in *Buffering*. - Flight or drive commands go over the same network. **This app never sends any**: it only asks for and receives video. -Whether a toy can be played here depends on its protocol. Toys that do offer RTSP or HTTP MJPEG play like any camera ([transports.md](transports.md)); others need a transport of their own, like [FLOW-UFO](#flow-ufo). +Whether a toy can be played here depends on its protocol. Toys that do offer RTSP or HTTP MJPEG play like any camera ([transports.md](transports.md)); others need a transport of their own, like [FLOW-UFO](#flow-ufo) or [HF / Jieli](#hf--jieli-toys). ## Connecting @@ -42,9 +43,9 @@ The Wi‑Fi name and the address the phone gets usually tell the family. The toy |---|---|---|---|---| | `FLOW-UFO-xxxxxx`, `FLOW_xxxxxx` | `192.168.169.x` | FLOW-UFO (BL60x camera; turbodrone calls it `wifi_uav`) | `flowufo://192.168.169.1` (UDP 8800 and 8801) | in testing ([#187](https://github.com/OpenIPC/decoder/issues/187)) | | `FLY-UFO-xxxxxx` and other cooingdv toys (KY UFO, RC UFO apps) | `192.168.1.x` | KY UFO RTSP | `rtsp://192.168.1.1:7070/webcam` (with the heartbeat to UDP 7099) | in testing ([#187](https://github.com/OpenIPC/decoder/issues/187)) | -| `HF_xxxxxxxx` | `192.168.80.x` | Jieli / HF | — | **not supported yet** | +| `HF_xxxxxxxx` | `192.168.80.x` | Jieli / HF (Jieli AC79 camera, vendor app *DVRunning 2*) | `jieli://192.168.80.1` (CTP on TCP 3333, video to the phone's UDP 6666) | test build 3, video path unverified ([#187](https://github.com/OpenIPC/decoder/issues/187)) | -When a `flowufo://` slot points at the phone itself, or at the network of another family, its status says so (for example *192.168.1.100 is this phone; the toy is usually 192.168.1.1* or *This network looks like a KY UFO RTSP toy: try rtsp://192.168.1.1:7070/webcam*). +When a `flowufo://` or `jieli://` slot points at the phone itself, or at the network of another family, its status says so (for example *192.168.1.100 is this phone; the toy is usually 192.168.1.1*, *This network looks like a KY UFO RTSP toy: try rtsp://192.168.1.1:7070/webcam* or, on a `192.168.80.x` network, *try jieli://192.168.80.1*). ## FLOW-UFO @@ -90,9 +91,42 @@ Their apps send a heartbeat, the two bytes `01 01`, to UDP port 7099 of the toy Some of these toys use the FLOW-UFO protocol instead (the KY UFO app has both); if RTSP shows nothing, try `flowufo://192.168.1.1`. +## HF / Jieli toys + +| | | +|---|---| +| Wi‑Fi name | `HF_xxxxxxxx` | +| Camera | Jieli AC79 (`AC79X_wifi_car_camera`); vendor app *DVRunning 2* (also *HF UFO*) | +| Toy address | `192.168.80.1`: control (CTP) on TCP port 3333; the video goes to UDP port 6666 **of the phone** | +| Slot URL | `jieli://192.168.80.1` (or `jieli://` alone) | +| Status | **test build 3**, not in the releases: the control part follows a capture of the vendor app, the video part is **unverified**; issue [#187](https://github.com/OpenIPC/decoder/issues/187) | + +What the app does, as a capture of the vendor app shows it ([#196](https://github.com/OpenIPC/decoder/issues/196)): + +- It connects to TCP port 3333 and talks **CTP**: each message is `CTP:`, the length of a topic name, the topic (e.g. `OPEN_RT_STREAM`), the length of a JSON text and the JSON. +- `APP_ACCESS` first; the toy answers with `KEEP_ALIVE_INTERVAL` (a timeout, 60 s), its battery and SD card state and `VIDEO_PARAM` (the size, e.g. 640×480 at 25 fps). Then `DATE_TIME` (the phone's time), a `CTP_KEEP_ALIVE` every 7.5 s (a quarter of a shorter timeout), and `OPEN_RT_STREAM` with the size from `VIDEO_PARAM` (640×480 at 25 if the toy sent none). When the slot stops: `CLOSE_RT_STREAM`. +- The video was **not** in the capture (PCAPdroid without root cannot see it). After turbodrone's research on the same Jieli SDK it is **RTP/JPEG** (payload type 26) or H.264 (96) to the phone's **UDP port 6666**. The app listens there and plays payload type 26 through its RTP/JPEG path and 96 through its H.264 path; a whole JPEG in a datagram without RTP is shown too. +- If nothing arrives on 6666 for 5 s after `OPEN_RT_STREAM`, the status says so and the app also listens on UDP 1234 (audio in the vendor SDK) and 2224 (its stream port), **only to log** what arrives there; it asks for the stream again every 10 s without video. +- **No flight command is ever sent**: no `FLIGHT_CONTROL` over CTP, nothing to the toy's UDP port 3456 (where the vendor app sends the sticks). +- Port 6666 can be used by one slot at a time: one `jieli://` slot in a grid. + +URL options (as with `udp://`, separated by `?` and `&`): + +| Option | Values | Default | What it does | +|---|---|---|---| +| `w`, `h`, `fps` | numbers | from `VIDEO_PARAM`, else 640, 480, 25 | the stream asked for in `OPEN_RT_STREAM` | +| `rtp` | a UDP port | `6666` | the local port the video is expected on | +| `format` | `0`, `1` | `0` | `0` asks for JPEG (as the vendor app does), `1` for H.264 | + +Examples: `jieli://192.168.80.1`, `jieli://192.168.80.1?w=1280&h=720`, `jieli://192.168.80.1?format=1`. + +Status texts while it connects: *Connecting to the toy (CTP 192.168.80.1:3333)*, *Stream requested, waiting for video on UDP 6666*, *No video on UDP 6666 yet: the toy accepted the request; see /log*, *Receiving from the toy on UDP 6666…*, *The toy's video stopped: waiting on UDP 6666*. The test build adds a `jieli` object to `/stats.json`: `state` (`connecting`, `ctp`, `requested`, `playing`, `quiet`), `target`, `network`, `ctp` (messages sent and received, keepalives and replies, the keepalive interval, the topics seen, the last 8 messages from the toy), `video_param`, `requested`, `rtp` (port, packets, video packets, frames, payload types, packets that are not RTP, the first packet in hex), `sources` (sender address:port → packets) and `diagnostic_ports`. The log has the CTP messages (JSON shortened) and the first packets on each port with sender, payload type and first bytes: exactly what a report needs. + +To test: install the test build, join the `HF_…` Wi‑Fi, [stay connected](#connecting), close the vendor app, put `jieli://192.168.80.1` into a slot. Wait about 30 s and [send a report](#what-to-send-in-a-report) in [#187](https://github.com/OpenIPC/decoder/issues/187), with or without a picture: `/stats.json` and `/log` tell what the toy really sends. + ## Credits -What the app does for FLOW-UFO and KY UFO toys was checked against [turbodrone](https://github.com/marshallrichards/turbodrone) (Apache-2.0), which flies these drones from a computer: its `wifi_uav` backend is the FLOW-UFO protocol and its `cooingdv` backend the KY UFO RTSP toys. The default quality parameters, the second port 8801 and its timing, the camera status bytes, leaving lagging frames out of the acknowledgement, the raster tables of `dqt=raster` and the 7099 heartbeat come from its code and research notes; no code of it is used. More open-source work on such toys is collected in [#196](https://github.com/OpenIPC/decoder/issues/196). +What the app does for FLOW-UFO and KY UFO toys was checked against [turbodrone](https://github.com/marshallrichards/turbodrone) (Apache-2.0), which flies these drones from a computer: its `wifi_uav` backend is the FLOW-UFO protocol and its `cooingdv` backend the KY UFO RTSP toys. The default quality parameters, the second port 8801 and its timing, the camera status bytes, leaving lagging frames out of the acknowledgement, the raster tables of `dqt=raster` and the 7099 heartbeat come from its code and research notes; no code of it is used. For the HF / Jieli toys its research notes on the same Jieli SDK (`rxdrone`, `cooingdv_jieli`: RTP/JPEG to UDP 6666, audio on 1234, the stream ports) describe the video part; the control part follows a PCAPdroid capture of the vendor app posted in [#196](https://github.com/OpenIPC/decoder/issues/196). More open-source work on such toys is collected in [#196](https://github.com/OpenIPC/decoder/issues/196). ## An unknown toy diff --git a/docs/transports.md b/docs/transports.md index 8d5634d2..375d8576 100644 --- a/docs/transports.md +++ b/docs/transports.md @@ -102,6 +102,8 @@ For example Majestic's `http://user:password@/mjpeg`: a `multipart/x-mix `flowufo://192.168.169.1:8800` (or `flowufo://` alone) plays the video of toy drones with a `FLOW-UFO-xxxxxx` Wi‑Fi access point. It exists only in a test build so far ([#187](https://github.com/OpenIPC/decoder/issues/187), PR [#188](https://github.com/OpenIPC/decoder/pull/188)), not in the releases. Port 8800 (the default) probes 8801 as well and keeps the port that answers. Options: `?qpara=<8 hex digits>` and `?dqt=zigzag|raster|q50`. Setup, options and status: [toys.md](toys.md#flow-ufo). +`jieli://192.168.80.1` (or `jieli://` alone) plays toy drones with an `HF_xxxxxxxx` Wi‑Fi (Jieli AC79 camera): CTP commands on TCP 3333 ask for the stream, which is expected as RTP/JPEG on the phone's UDP port 6666 (test build only, video path unverified). Options: `?w=`, `?h=`, `?fps=`, `?rtp=`, `?format=0|1`. See [toys.md](toys.md#hf--jieli-toys). + RTSP slots on `rtsp://:7070/webcam` (KY UFO and other cooingdv toys) also send the toy's UDP heartbeat `01 01` to port 7099 once a second; `?heartbeat=` turns it on for other RTSP URLs, `?heartbeat=off` off. See [toys.md](toys.md#ky-ufo-rtsp-toys). ## Codecs