diff --git a/.github/workflows/android-sender.yml b/.github/workflows/android-sender.yml new file mode 100644 index 0000000..e7ff5c7 --- /dev/null +++ b/.github/workflows/android-sender.yml @@ -0,0 +1,34 @@ +name: Android sender compile + +on: + workflow_dispatch: + pull_request: + paths: + - "android/sender/**" + - ".github/workflows/android-sender.yml" + push: + branches: + - main + paths: + - "android/sender/**" + - ".github/workflows/android-sender.yml" + +permissions: + contents: read + +jobs: + compile: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Check out repository + uses: actions/checkout@v6 + - name: Set up Java + uses: actions/setup-java@v5 + with: + distribution: temurin + java-version: "17" + - name: Test protocol core + run: ./android/sender/gradlew -p android/sender -PprotocolOnly=true :protocol:test --no-daemon + - name: Compile debug APK + run: ./android/sender/gradlew -p android/sender :app:assembleDebug --no-daemon diff --git a/android/sender/.gitignore b/android/sender/.gitignore new file mode 100644 index 0000000..208cf4f --- /dev/null +++ b/android/sender/.gitignore @@ -0,0 +1,3 @@ +.gradle/ +**/build/ +local.properties diff --git a/android/sender/README.md b/android/sender/README.md new file mode 100644 index 0000000..cf9a878 --- /dev/null +++ b/android/sender/README.md @@ -0,0 +1,38 @@ +# Android Screen Sender + +This source-only P1 slice lets an Android 14/API 34+ device publish a 720p30 +screen track to at most two Web Viewers assigned as direct children by the +existing `screener-v2` controller. + +## Build + +Install JDK 17 and Android SDK 35, then run: + +```text +./gradlew :protocol:test :app:assembleDebug +``` + +The debug APK is written under `app/build/outputs/apk/debug/`. CI runs the same +protocol tests and debug compile. No release signing or package publication is +configured. + +## Use + +Open the app, enter the HTTPS Screener base URL and site-access password, then +choose a screen or app from Android's system picker. The app keeps credentials +and room material in memory only. The displayed private invite can be opened by +an unchanged Web Viewer. + +The foreground notification and system capture indicator remain visible for the +whole session. Stopping from the app or system picker closes capture, signaling, +and every PeerConnection. + +## First-Slice Boundary + +- Video only: no playback-capture audio or A/V synchronization yet. +- Hardware VP8/H.264 encoders only; there is no software encoder fallback. +- At most two STUN-only direct-child edges. The server remains the route owner. +- SFU publication and selected-edge TURN are not implemented. The client reports + their attempts as bounded route failures instead of claiming support. +- Rotation resize, background rebuild, Wi-Fi/cellular reselection, iOS, release + signing, and device/performance matrices remain later work. diff --git a/android/sender/app/build.gradle.kts b/android/sender/app/build.gradle.kts new file mode 100644 index 0000000..da26a2f --- /dev/null +++ b/android/sender/app/build.gradle.kts @@ -0,0 +1,31 @@ +plugins { + id("com.android.application") + id("org.jetbrains.kotlin.android") +} + +android { + namespace = "icu.bonfire.screener.sender" + compileSdk = 35 + + defaultConfig { + applicationId = "icu.bonfire.screener.sender" + minSdk = 34 + targetSdk = 34 + versionCode = 1 + versionName = "0.1.0" + } + + compileOptions { + sourceCompatibility = JavaVersion.VERSION_17 + targetCompatibility = JavaVersion.VERSION_17 + } + kotlinOptions { + jvmTarget = "17" + } +} + +dependencies { + implementation(project(":protocol")) + implementation("com.squareup.okhttp3:okhttp:5.3.0") + implementation("io.github.webrtc-sdk:android:144.7559.12") +} diff --git a/android/sender/app/src/main/AndroidManifest.xml b/android/sender/app/src/main/AndroidManifest.xml new file mode 100644 index 0000000..4d8ca90 --- /dev/null +++ b/android/sender/app/src/main/AndroidManifest.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + diff --git a/android/sender/app/src/main/java/icu/bonfire/screener/sender/CaptureEngine.kt b/android/sender/app/src/main/java/icu/bonfire/screener/sender/CaptureEngine.kt new file mode 100644 index 0000000..59bf493 --- /dev/null +++ b/android/sender/app/src/main/java/icu/bonfire/screener/sender/CaptureEngine.kt @@ -0,0 +1,108 @@ +package icu.bonfire.screener.sender + +import android.content.Context +import android.content.Intent +import android.media.projection.MediaProjection +import org.webrtc.DefaultVideoDecoderFactory +import org.webrtc.EglBase +import org.webrtc.HardwareVideoEncoderFactory +import org.webrtc.PeerConnectionFactory +import org.webrtc.ScreenCapturerAndroid +import org.webrtc.SurfaceTextureHelper +import org.webrtc.VideoCodecInfo +import org.webrtc.VideoEncoder +import org.webrtc.VideoEncoderFactory +import org.webrtc.VideoSource +import org.webrtc.VideoTrack +import java.util.concurrent.atomic.AtomicBoolean + +class CaptureEngine private constructor( + context: Context, + permissionData: Intent, + onProjectionStopped: () -> Unit, +) : AutoCloseable { + private lateinit var egl: EglBase + lateinit var factory: PeerConnectionFactory + private set + lateinit var track: VideoTrack + private set + private lateinit var capturer: ScreenCapturerAndroid + private lateinit var source: VideoSource + private lateinit var helper: SurfaceTextureHelper + private var closed = false + + init { + try { + initializeOnce(context) + egl = EglBase.create() + val encoderFactory = HardwareOnlyEncoderFactory(egl.eglBaseContext) + if (encoderFactory.supportedCodecs.isEmpty()) error("No hardware VP8/H264 encoder is available") + factory = PeerConnectionFactory.builder() + .setVideoEncoderFactory(encoderFactory) + .setVideoDecoderFactory(DefaultVideoDecoderFactory(egl.eglBaseContext)) + .createPeerConnectionFactory() + source = factory.createVideoSource(true) + helper = SurfaceTextureHelper.create("ScreenerCapture", egl.eglBaseContext) + capturer = ScreenCapturerAndroid(permissionData, object : MediaProjection.Callback() { + override fun onStop() = onProjectionStopped() + }) + capturer.initialize(helper, context.applicationContext, source.capturerObserver) + capturer.startCapture(WIDTH, HEIGHT, FPS) + track = factory.createVideoTrack("screener-screen", source) + track.setEnabled(true) + } catch (error: Throwable) { + release() + throw error + } + } + + override fun close() = release() + + private fun release() { + if (closed) return + closed = true + if (::capturer.isInitialized) runCatching { capturer.stopCapture() } + if (::capturer.isInitialized) runCatching { capturer.dispose() } + if (::track.isInitialized) runCatching { track.dispose() } + if (::source.isInitialized) runCatching { source.dispose() } + if (::helper.isInitialized) runCatching { helper.dispose() } + if (::factory.isInitialized) runCatching { factory.dispose() } + if (::egl.isInitialized) runCatching { egl.release() } + } + + private class HardwareOnlyEncoderFactory(context: EglBase.Context) : VideoEncoderFactory { + private val delegate = HardwareVideoEncoderFactory(context, true, true) + override fun createEncoder(info: VideoCodecInfo): VideoEncoder? = + if (allowed(info)) delegate.createEncoder(info) else null + + override fun getSupportedCodecs(): Array = + delegate.supportedCodecs.filter(::allowed).toTypedArray() + + private fun allowed(info: VideoCodecInfo): Boolean = + info.name.equals("VP8", true) || info.name.equals("H264", true) + } + + companion object { + const val WIDTH = 1280 + const val HEIGHT = 720 + const val FPS = 30 + private val initialized = AtomicBoolean() + + fun create(context: Context, permissionData: Intent, onProjectionStopped: () -> Unit) = + CaptureEngine(context, permissionData, onProjectionStopped) + + private fun initializeOnce(context: Context) { + if (initialized.compareAndSet(false, true)) { + try { + PeerConnectionFactory.initialize( + PeerConnectionFactory.InitializationOptions.builder(context.applicationContext) + .createInitializationOptions(), + ) + } catch (error: Throwable) { + initialized.set(false) + throw error + } + } + } + } +} diff --git a/android/sender/app/src/main/java/icu/bonfire/screener/sender/HostPeer.kt b/android/sender/app/src/main/java/icu/bonfire/screener/sender/HostPeer.kt new file mode 100644 index 0000000..2064b0e --- /dev/null +++ b/android/sender/app/src/main/java/icu/bonfire/screener/sender/HostPeer.kt @@ -0,0 +1,158 @@ +package icu.bonfire.screener.sender + +import icu.bonfire.screener.protocol.IceConfig +import icu.bonfire.screener.protocol.SignalPayload +import icu.bonfire.screener.protocol.Wire +import icu.bonfire.screener.protocol.IceCandidate as WireCandidate +import org.webrtc.DataChannel +import org.webrtc.IceCandidate +import org.webrtc.MediaConstraints +import org.webrtc.MediaStream +import org.webrtc.PeerConnection +import org.webrtc.PeerConnectionFactory +import org.webrtc.RtpReceiver +import org.webrtc.SdpObserver +import org.webrtc.SessionDescription +import org.webrtc.VideoTrack +import java.util.concurrent.Executor + +class HostPeer( + private val peerId: String, + val connectionId: String, + iceConfig: IceConfig, + factory: PeerConnectionFactory, + track: VideoTrack, + private val serial: Executor, + private val send: (String) -> Boolean, + private val status: (String) -> Unit, +) : AutoCloseable { + private val pending = ArrayDeque() + @Volatile + private var closed = false + private val connection: PeerConnection + + init { + val servers = iceConfig.iceServers.map { PeerConnection.IceServer.builder(it.urls).createIceServer() } + val config = PeerConnection.RTCConfiguration(servers).apply { + sdpSemantics = PeerConnection.SdpSemantics.UNIFIED_PLAN + tcpCandidatePolicy = PeerConnection.TcpCandidatePolicy.DISABLED + continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY + } + connection = factory.createPeerConnection(config, Observer()) ?: error("Could not create PeerConnection") + try { + val sender = connection.addTrack(track, listOf("screener")) ?: error("Could not add screen track") + sender.parameters.let { parameters -> + parameters.encodings.forEach { it.maxBitrateBps = 3_000_000 } + if (!sender.setParameters(parameters)) error("Could not set video bitrate") + } + } catch (error: Throwable) { + closed = true + runCatching { connection.dispose() } + throw error + } + } + + fun start() = createOffer(false) + + fun restartIce() { + if (closed) return + connection.restartIce() + createOffer(true) + } + + fun accept(payload: SignalPayload) { + if (closed || payload.connectionId != connectionId) return + when (payload) { + is SignalPayload.Description -> { + val description = SessionDescription(SessionDescription.Type.ANSWER, payload.description.sdp) + connection.setRemoteDescription(object : SimpleSdpObserver() { + override fun onSetSuccess() = dispatch { flushCandidates() } + override fun onSetFailure(error: String?) = status("Viewer answer was rejected") + }, description) + } + is SignalPayload.Candidate -> { + if (connection.remoteDescription == null) { + if (pending.size < MAX_PENDING) pending += payload.candidate + } else { + addCandidate(payload.candidate) + } + } + } + } + + override fun close() { + if (closed) return + closed = true + pending.clear() + connection.dispose() + } + + private fun createOffer(restart: Boolean) { + connection.createOffer(object : SimpleSdpObserver() { + override fun onCreateSuccess(description: SessionDescription?) { + if (description == null || closed) return + connection.setLocalDescription(object : SimpleSdpObserver() { + override fun onSetSuccess() { + if (!closed && !send(Wire.description(peerId, connectionId, "offer", description.description))) { + status("Signaling is temporarily unavailable") + } + } + override fun onSetFailure(error: String?) = status("Could not apply the screen offer") + }, description) + } + override fun onCreateFailure(error: String?) = status(if (restart) "ICE restart failed" else "Viewer offer failed") + }, MediaConstraints()) + } + + private fun flushCandidates() { + while (!closed && pending.isNotEmpty()) addCandidate(pending.removeFirst()) + } + + private fun dispatch(block: () -> Unit) { + runCatching { serial.execute(block) } + } + + private fun addCandidate(candidate: WireCandidate?) { + if (candidate == null) return + connection.addIceCandidate( + IceCandidate(candidate.sdpMid ?: "", candidate.sdpMLineIndex ?: 0, candidate.candidate), + ) + } + + private inner class Observer : PeerConnection.Observer { + override fun onIceCandidate(candidate: IceCandidate) { + send(Wire.candidate(peerId, connectionId, WireCandidate( + candidate.sdp, + candidate.sdpMid, + candidate.sdpMLineIndex, + null, + ))) + } + override fun onIceGatheringChange(state: PeerConnection.IceGatheringState) { + if (state == PeerConnection.IceGatheringState.COMPLETE) send(Wire.candidate(peerId, connectionId, null)) + } + override fun onConnectionChange(state: PeerConnection.PeerConnectionState) { + if (state == PeerConnection.PeerConnectionState.FAILED) status("A viewer connection failed") + } + override fun onSignalingChange(state: PeerConnection.SignalingState) = Unit + override fun onIceConnectionChange(state: PeerConnection.IceConnectionState) = Unit + override fun onIceConnectionReceivingChange(receiving: Boolean) = Unit + override fun onIceCandidatesRemoved(candidates: Array) = Unit + override fun onAddStream(stream: MediaStream) = Unit + override fun onRemoveStream(stream: MediaStream) = Unit + override fun onDataChannel(channel: DataChannel) = Unit + override fun onRenegotiationNeeded() = Unit + override fun onAddTrack(receiver: RtpReceiver, streams: Array) = Unit + } + + private open class SimpleSdpObserver : SdpObserver { + override fun onCreateSuccess(description: SessionDescription?) = Unit + override fun onSetSuccess() = Unit + override fun onCreateFailure(error: String?) = Unit + override fun onSetFailure(error: String?) = Unit + } + + companion object { + private const val MAX_PENDING = 64 + } +} diff --git a/android/sender/app/src/main/java/icu/bonfire/screener/sender/MainActivity.kt b/android/sender/app/src/main/java/icu/bonfire/screener/sender/MainActivity.kt new file mode 100644 index 0000000..64a3ca4 --- /dev/null +++ b/android/sender/app/src/main/java/icu/bonfire/screener/sender/MainActivity.kt @@ -0,0 +1,105 @@ +package icu.bonfire.screener.sender + +import android.app.Activity +import android.content.ClipboardManager +import android.content.Context +import android.content.Intent +import android.media.projection.MediaProjectionManager +import android.os.Bundle +import android.os.Handler +import android.os.Looper +import android.os.ResultReceiver +import android.widget.Button +import android.widget.EditText +import android.widget.TextView + +class MainActivity : Activity() { + private lateinit var serverUrl: EditText + private lateinit var sitePassword: EditText + private lateinit var startButton: Button + private lateinit var stopButton: Button + private lateinit var statusText: TextView + private lateinit var inviteText: TextView + private var pendingServer = "" + private var pendingPassword = "" + + private val receiver = object : ResultReceiver(Handler(Looper.getMainLooper())) { + override fun onReceiveResult(resultCode: Int, resultData: Bundle?) { + statusText.text = resultData?.getString(ProjectionService.RESULT_STATUS) ?: "Sharing stopped" + resultData?.getString(ProjectionService.RESULT_INVITE)?.let { + inviteText.text = it + inviteText.setOnClickListener { _ -> + getSystemService(ClipboardManager::class.java).setPrimaryClip( + android.content.ClipData.newPlainText("Screener invite", it), + ) + } + } + val active = resultCode == ProjectionService.RESULT_ACTIVE + startButton.isEnabled = !active + stopButton.isEnabled = active + if (!active) clearInvite() + } + } + + override fun onCreate(savedInstanceState: Bundle?) { + super.onCreate(savedInstanceState) + setContentView(R.layout.activity_main) + serverUrl = findViewById(R.id.server_url) + sitePassword = findViewById(R.id.site_password) + startButton = findViewById(R.id.start_button) + stopButton = findViewById(R.id.stop_button) + statusText = findViewById(R.id.status_text) + inviteText = findViewById(R.id.invite_text) + + startButton.setOnClickListener { requestProjection() } + stopButton.setOnClickListener { + startService(Intent(this, ProjectionService::class.java).setAction(ProjectionService.ACTION_STOP)) + clearInvite() + statusText.text = "Stopping" + } + } + + private fun requestProjection() { + pendingServer = serverUrl.text.toString().trim().trimEnd('/') + pendingPassword = sitePassword.text.toString() + clearInvite() + if (!pendingServer.startsWith("https://")) { + statusText.text = "An HTTPS server URL is required" + return + } + val manager = getSystemService(MediaProjectionManager::class.java) + startActivityForResult(manager.createScreenCaptureIntent(), REQUEST_PROJECTION) + } + + @Deprecated("The platform projection picker still returns through this callback") + override fun onActivityResult(requestCode: Int, resultCode: Int, data: Intent?) { + super.onActivityResult(requestCode, resultCode, data) + if (requestCode != REQUEST_PROJECTION) return + if (resultCode != RESULT_OK || data == null) { + statusText.text = "Screen sharing was cancelled" + return + } + startButton.isEnabled = false + stopButton.isEnabled = true + statusText.text = "Starting" + val service = Intent(this, ProjectionService::class.java).apply { + action = ProjectionService.ACTION_START + putExtra(ProjectionService.EXTRA_SERVER_URL, pendingServer) + putExtra(ProjectionService.EXTRA_SITE_PASSWORD, pendingPassword) + putExtra(ProjectionService.EXTRA_PERMISSION_DATA, data) + putExtra(ProjectionService.EXTRA_RECEIVER, receiver) + } + startForegroundService(service) + sitePassword.text.clear() + pendingPassword = "" + } + + private fun clearInvite() { + inviteText.text = "" + inviteText.setOnClickListener(null) + } + + companion object { + private const val REQUEST_PROJECTION = 40 + } +} diff --git a/android/sender/app/src/main/java/icu/bonfire/screener/sender/ProjectionService.kt b/android/sender/app/src/main/java/icu/bonfire/screener/sender/ProjectionService.kt new file mode 100644 index 0000000..ea70906 --- /dev/null +++ b/android/sender/app/src/main/java/icu/bonfire/screener/sender/ProjectionService.kt @@ -0,0 +1,164 @@ +package icu.bonfire.screener.sender + +import android.app.Notification +import android.app.NotificationChannel +import android.app.NotificationManager +import android.app.PendingIntent +import android.app.Service +import android.content.Intent +import android.content.pm.ServiceInfo +import android.os.Bundle +import android.os.IBinder +import android.os.ResultReceiver +import java.util.concurrent.Executors +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.atomic.AtomicReference + +class ProjectionService : Service() { + private val worker = Executors.newSingleThreadExecutor() + private val state = AtomicReference(State.IDLE) + private var receiver: ResultReceiver? = null + private var capture: CaptureEngine? = null + @Volatile + private var session: RemoteSession? = null + + override fun onCreate() { + super.onCreate() + getSystemService(NotificationManager::class.java).createNotificationChannel( + NotificationChannel(CHANNEL_ID, "Screen sharing", NotificationManager.IMPORTANCE_LOW), + ) + } + + override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int { + if (intent?.action == ACTION_STOP) { + finish("Sharing stopped") + return START_NOT_STICKY + } + if (intent?.action != ACTION_START || !state.compareAndSet(State.IDLE, State.STARTING)) { + return START_NOT_STICKY + } + receiver = intent.getParcelableExtra(EXTRA_RECEIVER, ResultReceiver::class.java) + startForeground( + NOTIFICATION_ID, + notification("Starting screen sharing"), + ServiceInfo.FOREGROUND_SERVICE_TYPE_MEDIA_PROJECTION, + ) + val server = intent.getStringExtra(EXTRA_SERVER_URL) + val password = intent.getStringExtra(EXTRA_SITE_PASSWORD) + val permission = intent.getParcelableExtra(EXTRA_PERMISSION_DATA, Intent::class.java) + if (server == null || password == null || permission == null) { + finish("Screen sharing could not start") + return START_NOT_STICKY + } + sendResult(RESULT_ACTIVE, "Starting") + schedule { + try { + val engine = CaptureEngine.create(applicationContext, permission) { + finish("The system stopped screen sharing") + } + if (state.get() != State.STARTING) { + engine.close() + return@schedule + } + capture = engine + val remote = RemoteSession(server, password, engine.factory, engine.track, worker, object : RemoteSession.Events { + override fun onRoom(inviteUrl: String) { + if (state.get() == State.STOPPING) return + sendResult(RESULT_ACTIVE, "Room ready; hardware codec requested", inviteUrl) + getSystemService(NotificationManager::class.java) + .notify(NOTIFICATION_ID, notification("Screen sharing is ready")) + } + + override fun onStatus(message: String) { + if (state.get() != State.STOPPING) sendResult(RESULT_ACTIVE, message) + } + override fun onFatal(message: String) = finish(message) + }) + session = remote + if (!state.compareAndSet(State.STARTING, State.RUNNING)) return@schedule + remote.start() + } catch (_: Exception) { + finish("Screen sharing could not start") + } + } + return START_NOT_STICKY + } + + override fun onDestroy() { + state.set(State.STOPPING) + session?.cancelPending() + schedule { + teardown() + worker.shutdown() + } + super.onDestroy() + } + + override fun onBind(intent: Intent?): IBinder? = null + + private fun finish(message: String) { + if (state.getAndSet(State.STOPPING) == State.STOPPING) return + session?.cancelPending() + sendResult(RESULT_STOPPED, message) + schedule { + teardown() + stopForeground(STOP_FOREGROUND_REMOVE) + stopSelf() + } + } + + private fun teardown() { + session?.close() + session = null + capture?.close() + capture = null + } + + private fun schedule(block: () -> Unit) { + try { + worker.execute(block) + } catch (_: RejectedExecutionException) { + state.set(State.STOPPING) + } + } + + private fun sendResult(code: Int, status: String, invite: String? = null) { + receiver?.send(code, Bundle().apply { + putString(RESULT_STATUS, status) + invite?.let { putString(RESULT_INVITE, it) } + }) + } + + private fun notification(text: String): Notification { + val activity = PendingIntent.getActivity( + this, + 0, + Intent(this, MainActivity::class.java), + PendingIntent.FLAG_IMMUTABLE or PendingIntent.FLAG_UPDATE_CURRENT, + ) + return Notification.Builder(this, CHANNEL_ID) + .setSmallIcon(android.R.drawable.presence_video_online) + .setContentTitle("Screener Sender") + .setContentText(text) + .setContentIntent(activity) + .setOngoing(true) + .build() + } + + companion object { + const val ACTION_START = "icu.bonfire.screener.sender.START" + const val ACTION_STOP = "icu.bonfire.screener.sender.STOP" + const val EXTRA_SERVER_URL = "serverUrl" + const val EXTRA_SITE_PASSWORD = "sitePassword" + const val EXTRA_PERMISSION_DATA = "projectionPermission" + const val EXTRA_RECEIVER = "receiver" + const val RESULT_STATUS = "status" + const val RESULT_INVITE = "invite" + const val RESULT_STOPPED = 0 + const val RESULT_ACTIVE = 1 + private const val CHANNEL_ID = "screen-sharing" + private const val NOTIFICATION_ID = 70 + } + + private enum class State { IDLE, STARTING, RUNNING, STOPPING } +} diff --git a/android/sender/app/src/main/java/icu/bonfire/screener/sender/RemoteSession.kt b/android/sender/app/src/main/java/icu/bonfire/screener/sender/RemoteSession.kt new file mode 100644 index 0000000..1f68a9f --- /dev/null +++ b/android/sender/app/src/main/java/icu/bonfire/screener/sender/RemoteSession.kt @@ -0,0 +1,266 @@ +package icu.bonfire.screener.sender + +import icu.bonfire.screener.protocol.HostRoute +import icu.bonfire.screener.protocol.ProtocolException +import icu.bonfire.screener.protocol.RouteAction +import icu.bonfire.screener.protocol.ServerEvent +import icu.bonfire.screener.protocol.Wire +import okhttp3.Call +import okhttp3.Cookie +import okhttp3.CookieJar +import okhttp3.HttpUrl +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import okhttp3.MediaType.Companion.toMediaType +import okhttp3.RequestBody.Companion.toRequestBody +import okhttp3.HttpUrl.Companion.toHttpUrl +import org.webrtc.PeerConnectionFactory +import org.webrtc.VideoTrack +import java.util.UUID +import java.util.concurrent.Executor +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference + +class RemoteSession( + serverUrl: String, + private var sitePassword: String, + private val factory: PeerConnectionFactory, + private val track: VideoTrack, + private val serial: Executor, + private val events: Events, +) : AutoCloseable { + interface Events { + fun onRoom(inviteUrl: String) + fun onStatus(message: String) + fun onFatal(message: String) + } + + private val baseUrl = serverUrl.toHttpUrl() + private val cookieJar = MemoryCookieJar() + private val client = OkHttpClient.Builder().cookieJar(cookieJar).followRedirects(false).build() + private val route = HostRoute() + private val peers = linkedMapOf() + private val stopped = AtomicBoolean() + private val closed = AtomicBoolean() + private val activeCall = AtomicReference() + @Volatile + private var socket: WebSocket? = null + private var room: Room? = null + private var iceConfig: icu.bonfire.screener.protocol.IceConfig? = null + + init { + require( + baseUrl.isHttps && baseUrl.query == null && baseUrl.fragment == null && + baseUrl.username.isEmpty() && baseUrl.password.isEmpty(), + ) + } + + fun start() { + try { + authenticateSite() + } finally { + sitePassword = "" + } + if (stopped.get()) return + val created = createRoom() + if (stopped.get()) return + room = Room(created.roomId, created.hostToken, opaqueId(), opaqueId()) + events.onRoom(created.inviteUrl) + connectSignal() + } + + override fun close() { + if (!closed.compareAndSet(false, true)) return + stopped.set(true) + activeCall.get()?.cancel() + socket?.send(Wire.abandonRoom()) + socket?.close(1000, "sender stopped") + socket = null + peers.values.forEach(HostPeer::close) + peers.clear() + client.dispatcher.executorService.shutdown() + client.connectionPool.evictAll() + } + + fun cancelPending() { + stopped.set(true) + activeCall.get()?.cancel() + socket?.cancel() + } + + private fun authenticateSite() { + val request = Request.Builder() + .url(endpoint(SITE_ACCESS_PATH)) + .header("Accept", "application/json") + .header("Authorization", "Bearer ${sitePassword.trim()}") + .post("".toRequestBody(null)) + .build() + execute(request, "Site access was denied") + } + + private fun createRoom(): icu.bonfire.screener.protocol.CreatedRoom { + val body = """{"viewerPolicy":"private-link","hostClaimTtlSeconds":300}""" + val request = Request.Builder() + .url(endpoint(ROOMS_PATH)) + .header("Accept", "application/json") + .header("Content-Type", "application/json") + .header("Origin", origin()) + .post(body.toRequestBody("application/json".toMediaType())) + .build() + return Wire.decodeCreatedRoom(execute(request, "Room creation failed")) + } + + private fun execute(request: Request, failure: String): String { + val call = client.newCall(request) + check(activeCall.compareAndSet(null, call)) + try { + if (stopped.get()) call.cancel() + return call.execute().use { response -> + val text = response.body.string() + if (!response.isSuccessful || text.toByteArray().size > Wire.MAX_HTTP_BYTES) error(failure) + text + } + } finally { + activeCall.compareAndSet(call, null) + } + } + + private fun connectSignal() { + val wsUrl = endpoint(SIGNAL_PATH).toString().replaceFirst("https://", "wss://") + val cookies = cookieJar.loadForRequest(baseUrl).joinToString("; ") { "${it.name}=${it.value}" } + val request = Request.Builder().url(wsUrl).header("Origin", origin()).apply { + if (cookies.isNotEmpty()) header("Cookie", cookies) + }.build() + socket = client.newWebSocket(request, object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: Response) { + dispatch { + if (stopped.get()) return@dispatch + val identity = room ?: return@dispatch + webSocket.send(Wire.authenticate(identity.id, identity.token, identity.clientId, identity.shareGeneration)) + } + } + + override fun onMessage(webSocket: WebSocket, text: String) { + dispatch { if (!stopped.get()) accept(text) } + } + + override fun onFailure(webSocket: WebSocket, error: Throwable, response: Response?) { + fatal("Signaling connection failed") + } + + override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { + if (!stopped.get()) fatal("Signaling connection closed") + } + }) + } + + private fun accept(text: String) { + val event = try { + Wire.decodeServerEvent(text) + } catch (_: ProtocolException) { + fatal("The server protocol does not match this sender") + return + } + when (event) { + is ServerEvent.Authenticated -> { + iceConfig = event.iceConfig + val assignment = event.routeAssignment + val revision = event.routeRevision + if (assignment == null || revision == null) { + fatal("This sender requires peer-assisted media") + return + } + socket?.send(Wire.qualitySettings()) + apply(route.authoritative(event.peerId, revision, assignment)) + events.onStatus("Connected; ${peers.size} viewer edge(s)") + } + is ServerEvent.RouteUpdate -> apply(route.update(event)) + is ServerEvent.SelectedSfuIngress -> apply(route.selectedSfuIngress(event)) + is ServerEvent.Signal -> { + if (route.isAssignedChild(event.fromPeerId)) peers[event.fromPeerId]?.accept(event.payload) + } + is ServerEvent.RestartRequest -> recover(event) + is ServerEvent.RoomClosed -> fatal(if (event.reason == "expired") "Room expired" else "Room closed") + is ServerEvent.Error -> { + if (event.code in FATAL_ERRORS) fatal("The server rejected this sender") + else events.onStatus("A media route is unavailable") + } + is ServerEvent.Ignored -> Unit + } + } + + private fun apply(actions: List) { + actions.forEach { + when (it) { + is RouteAction.ReconcileChildren -> reconcile(it.peerIds) + is RouteAction.Ready -> socket?.send(Wire.routeReady(it.revision, it.phase)) + is RouteAction.Failed -> socket?.send(Wire.routeFailed(it.revision, it.phase, it.connectionId)) + } + } + } + + private fun reconcile(peerIds: Collection) { + val desired = peerIds.distinct().take(MAX_EDGES).toSet() + (peers.keys - desired).forEach { peers.remove(it)?.close() } + desired.filterNot(peers::containsKey).forEach { peerId -> + val config = iceConfig ?: return@forEach + peers[peerId] = HostPeer(peerId, opaqueId(), config, factory, track, serial, ::send, events::onStatus).also { + it.start() + } + } + } + + private fun recover(event: ServerEvent.RestartRequest) { + val peer = peers[event.fromPeerId] ?: return + if (peer.connectionId != event.connectionId) return + if (event.rebuild) { + peer.close() + peers.remove(event.fromPeerId) + reconcile(routeChildren() + event.fromPeerId) + } else { + peer.restartIce() + } + } + + private fun routeChildren(): Set = peers.keys.toSet() + private fun send(message: String): Boolean = socket?.send(message) == true + private fun dispatch(block: () -> Unit) { + runCatching { serial.execute(block) } + } + private fun fatal(message: String) { + if (stopped.get()) return + events.onFatal(message) + } + + private fun endpoint(path: String): HttpUrl = baseUrl.newBuilder().encodedPath(path).query(null).fragment(null).build() + private fun origin(): String = baseUrl.newBuilder().encodedPath("/").query(null).fragment(null).build().toString().removeSuffix("/") + + private data class Room(val id: String, val token: String, val clientId: String, val shareGeneration: String) + + private class MemoryCookieJar : CookieJar { + private val cookies = mutableListOf() + override fun saveFromResponse(url: HttpUrl, values: List) = synchronized(cookies) { + values.forEach { value -> + cookies.removeAll { it.name == value.name && it.domain == value.domain && it.path == value.path } + if (value.expiresAt > System.currentTimeMillis()) cookies += value + } + } + + override fun loadForRequest(url: HttpUrl): List = synchronized(cookies) { + cookies.removeAll { it.expiresAt <= System.currentTimeMillis() } + cookies.filter { it.matches(url) } + } + } + + companion object { + private const val SITE_ACCESS_PATH = "/api/site-access" + private const val ROOMS_PATH = "/api/rooms" + private const val SIGNAL_PATH = "/signal" + private const val MAX_EDGES = 2 + private val FATAL_ERRORS = setOf("AUTH_REQUIRED", "INVALID_TOKEN", "ROOM_EXPIRED", "HOST_ALREADY_CONNECTED") + private fun opaqueId(): String = UUID.randomUUID().toString() + } +} diff --git a/android/sender/app/src/main/res/layout/activity_main.xml b/android/sender/app/src/main/res/layout/activity_main.xml new file mode 100644 index 0000000..fe1b2a3 --- /dev/null +++ b/android/sender/app/src/main/res/layout/activity_main.xml @@ -0,0 +1,80 @@ + + + + + + + + + + + +