|
1 | 1 | package com.squareup.workflow1.traceviewer.util |
2 | 2 |
|
3 | 3 | import kotlinx.coroutines.Dispatchers |
4 | | -import kotlinx.coroutines.channels.Channel |
| 4 | +import kotlinx.coroutines.awaitCancellation |
| 5 | +import kotlinx.coroutines.coroutineScope |
| 6 | +import kotlinx.coroutines.ensureActive |
| 7 | +import kotlinx.coroutines.launch |
| 8 | +import kotlinx.coroutines.runInterruptible |
5 | 9 | import kotlinx.coroutines.withContext |
| 10 | +import okio.IOException |
6 | 11 | import java.net.Socket |
7 | | -import java.net.SocketException |
8 | 12 |
|
9 | 13 | /** |
10 | | - * This is a client that can connect to any server socket that sends render pass data while using |
11 | | - * the Workflow framework. |
| 14 | + * Collects data from a server socket and serves them back to the caller via callback. |
12 | 15 | * |
13 | | - * [start] and [close] are idempotent commands, so this socket can only be started and closed once. |
| 16 | + * Two cases that are guaranteed to fail: |
| 17 | + * 1) The app is not running |
| 18 | + * 2) A reattempt at establishing socket connection without restarting the app |
14 | 19 | * |
15 | | - * Since this app is on JVM and the server is on Android, we use ADB to forward the port onto the socket. |
| 20 | + * @param onNewRenderPass is called from an arbitrary thread, so it is important to ensure that the |
| 21 | + * caller is thread safe |
16 | 22 | */ |
17 | | -internal class SocketClient { |
18 | | - private lateinit var socket: Socket |
19 | | - private var initialized = false |
20 | | - val renderPassChannel: Channel<String> = Channel(Channel.BUFFERED) |
21 | | - |
22 | | - /** |
23 | | - * We use any available ports on the host machine to connect to the emulator. |
24 | | - * |
25 | | - * `workflow-trace` is the name of the unix socket created, and since Android uses |
26 | | - * `LocalServerSocket` -- which creates a unix socket on the linux abstract namespace -- we use |
27 | | - * `localabstract:` to connect to it. |
28 | | - */ |
29 | | - fun open() { |
30 | | - if (initialized) { |
31 | | - return |
| 23 | +suspend fun pollSocket(onNewRenderPass: suspend (String) -> Unit) { |
| 24 | + withContext(Dispatchers.IO) { |
| 25 | + try { |
| 26 | + runForwardingPortThroughAdb { port -> |
| 27 | + Socket("localhost", port).useWithCancellation { socket -> |
| 28 | + val reader = socket.getInputStream().bufferedReader() |
| 29 | + do { |
| 30 | + ensureActive() |
| 31 | + val input = reader.readLine() |
| 32 | + if (input != null) { |
| 33 | + onNewRenderPass(input) |
| 34 | + } |
| 35 | + } while (input != null) |
| 36 | + } |
| 37 | + } |
| 38 | + } catch (e: IOException) { |
| 39 | + // NoOp |
32 | 40 | } |
33 | | - initialized = true |
34 | | - val process = ProcessBuilder( |
35 | | - "adb", "forward", "tcp:0", "localabstract:workflow-trace" |
36 | | - ).start() |
| 41 | + } |
| 42 | +} |
37 | 43 |
|
38 | | - // The adb forward command will output the port number it picks to connect. |
39 | | - process.waitFor() |
40 | | - val port = process.inputStream.bufferedReader().readText() |
41 | | - .trim().toInt() |
| 44 | +/** |
| 45 | + * Force [pollSocket] to exit with exception if the coroutine is cancelled. See comment below. |
| 46 | + */ |
| 47 | +private suspend fun Socket.useWithCancellation(block: suspend (Socket) -> Unit) { |
| 48 | + val socket = this |
| 49 | + coroutineScope { |
| 50 | + // This coroutine is responsible for forcibly closing the socket when the coroutine is |
| 51 | + // cancelled. This causes any code reading from the socket to throw a CancellationException. |
| 52 | + // We also need to explicitly cancel this coroutine if the block returns on its own, otherwise |
| 53 | + // the coroutineScope will never exit. |
| 54 | + val socketJob = launch { |
| 55 | + socket.use { |
| 56 | + awaitCancellation() |
| 57 | + } |
| 58 | + } |
42 | 59 |
|
43 | | - socket = Socket("localhost", port) |
| 60 | + block(socket) |
| 61 | + socketJob.cancel() |
44 | 62 | } |
| 63 | +} |
45 | 64 |
|
46 | | - fun close() { |
47 | | - if (!initialized) { |
48 | | - return |
49 | | - } |
50 | | - socket.close() |
| 65 | +/** |
| 66 | + * Call adb to setup a port forwarding to the server socket, and calls block with the allocated |
| 67 | + * port number if successful. |
| 68 | + * |
| 69 | + * If block throws or returns on finish, the port forwarding is removed via adb (best effort). |
| 70 | + */ |
| 71 | +@Suppress("BlockingMethodInNonBlockingContext") |
| 72 | +private suspend inline fun runForwardingPortThroughAdb(block: (port: Int) -> Unit) { |
| 73 | + val process = ProcessBuilder( |
| 74 | + "adb", "forward", "tcp:0", "localabstract:workflow-trace" |
| 75 | + ).start() |
| 76 | + |
| 77 | + // The adb forward command will output the port number it picks to connect. |
| 78 | + val forwardReturnCode = runInterruptible { |
| 79 | + process.waitFor() |
| 80 | + } |
| 81 | + if (forwardReturnCode != 0) { |
| 82 | + return |
51 | 83 | } |
52 | 84 |
|
53 | | - /** |
54 | | - * Polls the socket's input stream and sends the data into [renderPassChannel]. |
55 | | - * The caller should handle the scope of the coroutine that this function is called in. |
56 | | - * |
57 | | - * To better separate the responsibility of reading from the socket, we use a channel for the caller |
58 | | - * to handle parsing and amalgamating the render passes. |
59 | | - */ |
60 | | - suspend fun pollSocket() { |
61 | | - withContext(Dispatchers.IO) { |
62 | | - val reader = socket.getInputStream().bufferedReader() |
63 | | - reader.use { |
64 | | - try { |
65 | | - while (true) { |
66 | | - val input = reader.readLine() |
67 | | - renderPassChannel.trySend(input) |
68 | | - } |
69 | | - } catch (e: SocketException) { |
70 | | - e.printStackTrace() |
71 | | - } |
72 | | - } |
| 85 | + val port = process.inputStream.bufferedReader().readText() |
| 86 | + .trim().toInt() |
| 87 | + |
| 88 | + try { |
| 89 | + block(port) |
| 90 | + } finally { |
| 91 | + // We don't care if this fails since there's nothing we can do then anyway. It just means |
| 92 | + // there's an extra forward left open, but that's not a big deal. |
| 93 | + runCatching { |
| 94 | + ProcessBuilder( |
| 95 | + "adb", "forward", "--remove", "tcp:$port" |
| 96 | + ).start() |
73 | 97 | } |
74 | 98 | } |
75 | 99 | } |
0 commit comments