subrep-android/app/src/main/kotlin/com/honjimaku/subrep/Socket.kt

app/src/main/kotlin/com/honjimaku/subrep/Socket.kt (2822 bytes)
1 package com.honjimaku.subrep
2 
3 import okhttp3.OkHttpClient
4 import okhttp3.Request
5 import okhttp3.Response
6 import okhttp3.WebSocket
7 import okhttp3.WebSocketListener
8 import okio.ByteString.Companion.toByteString
9 import java.util.concurrent.Executors
10 import java.util.concurrent.TimeUnit
11 
12 /**
13  * One websocket that comes back on its own after a loss. The first message after each connect
14  * is the hello of the protocol, so the other side knows what follows.
15  */
16 class Socket(
17     private val client: OkHttpClient,
18     private val url: String,
19     private val hello: () -> String,
20     private val onText: (String) -> Unit,
21     /** True when the socket is open, false when it was lost. Not called for a failed connect. */
22     private val onState: (Boolean) -> Unit,
23 ) : WebSocketListener() {
24 
25     private val timer = Executors.newSingleThreadScheduledExecutor()
26 
27     @Volatile private var socket: WebSocket? = null
28     @Volatile private var wanted = false
29     @Volatile var open = false
30         private set
31     private var backoffMs = 1000L
32 
33     fun start() {
34         wanted = true
35         connect()
36     }
37 
38     fun stop() {
39         wanted = false
40         open = false
41         timer.shutdownNow()
42         socket?.close(1000, "stop")
43         socket = null
44     }
45 
46     fun send(text: String): Boolean = open && socket?.send(text) == true
47 
48     /**
49      * Sends the first [count] bytes of [bytes]. Dropped when the socket is not open, or when
50      * a slow connection holds more than a few seconds of sound: late captions help nobody.
51      */
52     fun send(bytes: ByteArray, count: Int): Boolean {
53         val s = socket ?: return false
54         if (!open || s.queueSize() > QUEUE_LIMIT) return false
55         return s.send(bytes.toByteString(0, count))
56     }
57 
58     private fun connect() {
59         if (!wanted) return
60         socket = client.newWebSocket(Request.Builder().url(url).build(), this)
61     }
62 
63     override fun onOpen(webSocket: WebSocket, response: Response) {
64         backoffMs = 1000
65         open = true
66         webSocket.send(hello())
67         onState(true)
68     }
69 
70     override fun onMessage(webSocket: WebSocket, text: String) = onText(text)
71 
72     override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) = lost()
73 
74     override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
75         webSocket.close(1000, null)
76     }
77 
78     override fun onClosed(webSocket: WebSocket, code: Int, reason: String) = lost()
79 
80     private fun lost() {
81         val was = open
82         open = false
83         if (was) onState(false)
84         if (!wanted) return
85         val wait = backoffMs
86         backoffMs = minOf(backoffMs * 2, 30_000)
87         runCatching { timer.schedule({ connect() }, wait, TimeUnit.MILLISECONDS) }
88     }
89 
90     private companion object {
91         const val QUEUE_LIMIT = 256 * 1024L
92     }
93 }