upApp

upApp/app/src/main/java/de/upapp/Tunnel.kt
Tunnel.kt
⎇
Raw
1package de.upapp
2
3import android.util.Log
4import java.io.Closeable
5import java.io.IOException
6import java.io.InputStream
7import java.io.OutputStream
8import java.net.SocketTimeoutException
9import java.util.concurrent.ConcurrentHashMap
10import java.util.concurrent.LinkedBlockingQueue
11import java.util.concurrent.TimeUnit
12import java.util.concurrent.atomic.AtomicInteger
13import kotlin.concurrent.thread
14
15internal const val TRANSMIT = 0
16internal const val OPEN = 1
17internal const val CLOSE = 2
18internal const val HEARTBEAT = 3
19private const val MAX_PAYLOAD = 0xFFFF
20// Values from maps+more.
21private const val HEARTBEAT_MS = 5_000L
22private const val SILENCE_MS = 15_000L
23
24internal class Segment(val port: Int, val type: Int, val data: ByteArray)
25
26internal fun frame(port: Int, type: Int, extra: ByteArray = ByteArray(0)) =
27 byteArrayOf(port.toByte(), (port shr 8).toByte(), type.toByte()) + extra
28
29internal fun u16(v: Int) = byteArrayOf(v.toByte(), (v shr 8).toByte())
30
31private fun u16(b: ByteArray, at: Int) = (b[at].toInt() and 0xFF) or ((b[at + 1].toInt() and 0xFF) shl 8)
32
33/** Returns the first segment in [b] and its encoded size, or null if [b] holds no complete segment. */
34internal fun decode(b: ByteArray): Pair<Segment, Int>? {
35 if (b.size < 3) return null
36 val port = u16(b, 0)
37 return when (val type = b[2].toInt() and 0xFF) {
38 CLOSE, HEARTBEAT -> Segment(port, type, ByteArray(0)) to 3
39 OPEN -> if (b.size < 5) null else Segment(port, type, b.copyOfRange(3, 5)) to 5
40 TRANSMIT -> {
41 if (b.size < 5) return null
42 val end = 5 + u16(b, 3)
43 if (b.size < end) null else Segment(port, type, b.copyOfRange(5, end)) to end
44 }
45 else -> throw IOException("Unknown segment type $type")
46 }
47}
48
49/** MIBBridge protocol: multiplexes virtual TCP connections to the car over one RFCOMM stream. */
50class Tunnel(private val input: InputStream, private val output: OutputStream) : Closeable {
51 private val channels = ConcurrentHashMap<Int, Channel>()
52 private val nextPort = AtomicInteger()
53 @Volatile var closed = false
54 private set
55 @Volatile private var lastReceived = System.currentTimeMillis()
56
57 init {
58 thread(isDaemon = true, name = "tunnel") { readLoop() }
59 thread(isDaemon = true, name = "heartbeat") { heartbeatLoop() }
60 }
61
62 /** Opens a connection to [remotePort] on the car. Reads fail after [timeoutMs] without data. */
63 fun open(remotePort: Int, timeoutMs: Long = 15_000): Channel {
64 if (closed) throw IOException("Bluetooth connection closed")
65 while (true) {
66 val ch = Channel(0xC000 + (nextPort.getAndIncrement() and 0x3FFF), timeoutMs)
67 if (channels.putIfAbsent(ch.port, ch) == null) {
68 send(frame(ch.port, OPEN, u16(remotePort)))
69 return ch
70 }
71 }
72 }
73
74 override fun close() {
75 if (closed) return
76 closed = true
77 channels.values.forEach { it.pipe.end() }
78 channels.clear()
79 runCatching { input.close() }
80 runCatching { output.close() }
81 }
82
83 private fun send(bytes: ByteArray) = synchronized(output) {
84 output.write(bytes)
85 output.flush()
86 }
87
88 private fun heartbeatLoop() {
89 try {
90 while (!closed) {
91 send(frame(0, HEARTBEAT))
92 Thread.sleep(HEARTBEAT_MS)
93 if (System.currentTimeMillis() - lastReceived > SILENCE_MS) {
94 Log.w(TAG, "No data from the car for ${SILENCE_MS / 1000} s")
95 break
96 }
97 }
98 } catch (e: IOException) {
99 if (!closed) Log.w(TAG, "Heartbeat failed", e)
100 } finally {
101 close()
102 }
103 }
104
105 private fun readLoop() {
106 val chunk = ByteArray(4096)
107 var buf = ByteArray(0)
108 try {
109 while (true) {
110 val n = input.read(chunk)
111 if (n < 0) break
112 lastReceived = System.currentTimeMillis()
113 buf += chunk.copyOf(n)
114 while (true) {
115 val (segment, size) = decode(buf) ?: break
116 buf = buf.copyOfRange(size, buf.size)
117 handle(segment)
118 }
119 }
120 } catch (e: IOException) {
121 if (!closed) Log.w(TAG, "Tunnel read failed", e)
122 } finally {
123 close()
124 }
125 }
126
127 private fun handle(s: Segment) {
128 when (s.type) {
129 HEARTBEAT -> {}
130 TRANSMIT -> channels[s.port]?.pipe?.put(s.data) ?: Log.w(TAG, "Data for unknown port ${s.port}")
131 CLOSE -> channels[s.port]?.onCarClose()
132 else -> Log.w(TAG, "Unexpected segment type ${s.type}")
133 }
134 }
135
136 inner class Channel(val port: Int, timeoutMs: Long) : Closeable {
137 internal val pipe = Pipe(timeoutMs)
138 @Volatile private var closedByCar = false
139 @Volatile private var closedByUs = false
140
141 val input: InputStream = pipe.input
142
143 val output: OutputStream = object : OutputStream() {
144 override fun write(b: Int) = write(byteArrayOf(b.toByte()), 0, 1)
145
146 override fun write(b: ByteArray, off: Int, len: Int) {
147 if (closedByCar || closed) throw IOException("Connection closed by the car")
148 var at = off
149 while (at < off + len) {
150 val n = minOf(MAX_PAYLOAD, off + len - at)
151 send(frame(port, TRANSMIT, u16(n) + b.copyOfRange(at, at + n)))
152 at += n
153 }
154 }
155 }
156
157 override fun close() {
158 if (closedByUs) return
159 closedByUs = true
160 pipe.end()
161 if (closedByCar) return
162 // The car acknowledges with its own CLOSE. onCarClose then unregisters the port.
163 runCatching { send(frame(port, CLOSE)) }.onFailure { channels.remove(port) }
164 }
165
166 internal fun onCarClose() {
167 if (!closedByUs) {
168 closedByCar = true
169 pipe.end()
170 runCatching { send(frame(port, CLOSE)) }
171 }
172 channels.remove(port)
173 }
174 }
175
176 private companion object {
177 const val TAG = "Tunnel"
178 }
179}
180
181/** In-memory byte stream between threads. Reads fail after [timeoutMs] without data. */
182internal class Pipe(private val timeoutMs: Long = Long.MAX_VALUE) {
183 private val queue = LinkedBlockingQueue<ByteArray>()
184 private var current = ByteArray(0)
185 private var pos = 0
186 private var ended = false
187
188 fun put(bytes: ByteArray) {
189 if (bytes.isNotEmpty()) queue.add(bytes)
190 }
191
192 fun end() {
193 queue.add(END)
194 }
195
196 val input: InputStream = object : InputStream() {
197 override fun read(): Int {
198 val b = ByteArray(1)
199 return if (read(b, 0, 1) < 0) -1 else b[0].toInt() and 0xFF
200 }
201
202 override fun read(b: ByteArray, off: Int, len: Int): Int {
203 if (len == 0) return 0
204 while (pos == current.size) {
205 if (ended) return -1
206 val next = queue.poll(timeoutMs, TimeUnit.MILLISECONDS)
207 ?: throw SocketTimeoutException("No answer from the car for ${timeoutMs / 1000} s")
208 ended = next === END
209 current = next
210 pos = 0
211 }
212 val n = minOf(len, current.size - pos)
213 current.copyInto(b, off, pos, pos + n)
214 pos += n
215 return n
216 }
217
218 override fun close() = end()
219 }
220
221 val output: OutputStream = object : OutputStream() {
222 override fun write(b: Int) = put(byteArrayOf(b.toByte()))
223 override fun write(b: ByteArray, off: Int, len: Int) = put(b.copyOfRange(off, off + len))
224 override fun close() = end()
225 }
226
227 private companion object {
228 val END = ByteArray(0)
229 }
230}
231