Android WebRTC DataChannel is always null and ICE candidates are not sent

Question

I'm trying to create an Android app that uses WebRTC data channels for data exchange. The data that I want to send is basic strings. Admittedly, this is my first time looking at WebRTC, and so I am a bit fuzzy on the details. My problem is that whenever I try to create a data channel, it is always null, and ICE candidate requests do not seem to be exchanged with the signalling server. I started from this example that creates a connection to exchange video between two devices and modified it to not exchange video but instead create a data channel.

I looked through a lot of other answers but the vast majority have to do with WebRTC in the browser, and data channel examples are rare to begin with. I also looked through the google chromium source code implementation of WebRTC in c++ to see if anything could be learned but had no luck.

My code is as follows

WebRtcActivity.kt

// imports omitted
class WebRtcActivity : AppCompatActivity() {

    private lateinit var rtcClient: RTCClient
    private lateinit var signallingClient: SignallingClient

    private val sdpObserver = object : AppSdpObserver() {
        override fun onCreateSuccess(p0: SessionDescription?) {
            super.onCreateSuccess(p0)
            signallingClient.send(p0)
            println("session description to string: " + p0.toString()) // prints
        }
    }

    
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_web_rtc)

        // camera permission check omitted
        onCameraPermissionGranted()
        
    }

    private fun onCameraPermissionGranted() {
        rtcClient = RTCClient(
                application,
                object : PeerConnectionObserver() {
                    override fun onIceCandidate(p0: IceCandidate?) {
                        super.onIceCandidate(p0)
                        signallingClient.send(p0)
                        rtcClient.addIceCandidate(p0)
                        println("ice candidate to string: " + p0.toString()) // does not print
                    }

                    override fun onDataChannel(p0: DataChannel?) {
                        super.onDataChannel(p0)
                    }

                }
        )

        signallingClient = SignallingClient(createSignallingClientListener())

        call_button.setOnClickListener { // on-screen button to initiate sending the offer
            rtcClient.call(sdpObserver)
        }
    }
    
    private fun createSignallingClientListener() = object : SignallingClientListener {
        override fun onConnectionEstablished() {
            println("connection established")
        }

        override fun onOfferReceived(description: SessionDescription) {
            rtcClient.onRemoteSessionReceived(description)
            rtcClient.answer(sdpObserver)
            println("offer received") // prints
        }
        
        override fun onAnswerReceived(description: SessionDescription) {
            rtcClient.onRemoteSessionReceived(description)
            println("answer received") // prints
        }

        override fun onIceCandidateReceived(iceCandidate: IceCandidate) {
            rtcClient.addIceCandidate(iceCandidate)
            println("signalling client ice candidate") // does not print
        }
    }

RTCClient.kt

// imports omitted
class RTCClient(
        context: Application,
        observer: PeerConnection.Observer
) {

    init {
        initPeerConnectionFactory(context)
    }

    private val iceServer = listOf(
            PeerConnection.IceServer.builder("stun:stun1.l.google.com:19302")
                    .createIceServer()
    )

    private val peerConnectionFactory by lazy { buildPeerConnectionFactory() }
    private val peerConnection by lazy { buildPeerConnection(observer) }

    private fun initPeerConnectionFactory(context: Application) {
        val options = PeerConnectionFactory.InitializationOptions.builder(context)
                .setEnableInternalTracer(true)
                .createInitializationOptions()
        PeerConnectionFactory.initialize(options)
    }

    private fun buildPeerConnectionFactory(): PeerConnectionFactory {
        return PeerConnectionFactory
                .builder()
                .setOptions(PeerConnectionFactory.Options().apply {
                    disableEncryption = true
                    disableNetworkMonitor = true
                })
                .createPeerConnectionFactory()
    }

    private fun buildPeerConnection(observer: PeerConnection.Observer) = peerConnectionFactory.createPeerConnection(
            iceServer,
            observer
    )

    private fun PeerConnection.call(sdpObserver: SdpObserver) {
        val constraints = MediaConstraints().apply {
            //mandatory.add(MediaConstraints.KeyValuePair("OfferToReceiveAudio", "true"))  // with this, the ICE candidate requests actually send, but I am not using audio so i do not think it needs to be here
            mandatory.add(MediaConstraints.KeyValuePair("DtlsSrtpKeyAgreement", "true")) // saw this constraint on an answer somewhere, I cannot tell if it is needed/or not
        }
        
        createOffer(object : SdpObserver by sdpObserver {
            override fun onCreateSuccess(desc: SessionDescription?) {

                setLocalDescription(object : SdpObserver {
                    override fun onSetFailure(p0: String?) {
                    }

                    override fun onSetSuccess() {
                    }

                    override fun onCreateSuccess(p0: SessionDescription?) {
                    }

                    override fun onCreateFailure(p0: String?) {
                    }
                }, desc)
                sdpObserver.onCreateSuccess(desc)
            }
        }, constraints)
    }

    private fun PeerConnection.answer(sdpObserver: SdpObserver) {
        val constraints = MediaConstraints().apply {
            //mandatory.add(MediaConstraints.KeyValuePair("OfferToReceiveAudio", "true")) // with this, the ICE candidate requests actually send, but I am not using audio so i do not think it needs to be here
            mandatory.add(MediaConstraints.KeyValuePair("DtlsSrtpKeyAgreement", "true"))
        }
        
        createAnswer(object : SdpObserver by sdpObserver {
            override fun onCreateSuccess(p0: SessionDescription?) {
                setLocalDescription(object : SdpObserver {
                    override fun onSetFailure(p0: String?) {
                    }

                    override fun onSetSuccess() {
                    }

                    override fun onCreateSuccess(p0: SessionDescription?) {
                    }

                    override fun onCreateFailure(p0: String?) {
                    }
                }, p0)
                sdpObserver.onCreateSuccess(p0)
            }
        }, constraints)

    }

    private fun createChannel(label: String, peerConnection: PeerConnection?) {
        try {
            val init = DataChannel.Init().apply {
                negotiated = true
                id = 0
            }
            val channel = peerConnection?.createDataChannel(label, init)
            // here, channel is null

            val channelObserver: DataChannelObserver = DataChannelObserver()
            channel?.registerObserver(channelObserver)
            
        } catch (e: Throwable) {
            e.printStackTrace()
            println("creating channel exception: $e")
        }
    }

    fun call(sdpObserver: SdpObserver) = peerConnection?.call(sdpObserver)

    fun answer(sdpObserver: SdpObserver) = peerConnection?.answer(sdpObserver)

    fun createDataChannel(label: String) = this.createChannel(label, peerConnection)

    fun onRemoteSessionReceived(sessionDescription: SessionDescription) {
        peerConnection?.setRemoteDescription(object : SdpObserver {
            override fun onSetFailure(p0: String?) {
            }

            override fun onSetSuccess() {
            }

            override fun onCreateSuccess(p0: SessionDescription?) {
            }

            override fun onCreateFailure(p0: String?) {
            }
        }, sessionDescription)
    }

    fun addIceCandidate(iceCandidate: IceCandidate?) {
        peerConnection?.addIceCandidate(iceCandidate)
    }
}

SignallingClient.kt

// imports omitted=
class SignallingClient(
    private val listener: SignallingClientListener
) : CoroutineScope {
    
    private val job = Job()

    private val gson = Gson()

    override val coroutineContext = Dispatchers.IO + job

    private val sendChannel = ConflatedBroadcastChannel<String>()

    init {
        connect()
    }
    
    private fun connect() = launch {
        val client = OkHttpClient()

        val request = Request.Builder().url("wss://random_ngrok_url").build() // here, I use ngrok since android has a problem with sending cleartext to a non-encrypted location. i also have the signalling server hosted on heroku, and it works the same

        val wsListener = CustomWebSocketListener(listener)

        val ws: WebSocket = client.newWebSocket(request, wsListener)
        
        listener.onConnectionEstablished()
        
        val sendData = sendChannel.openSubscription()
        
        try {
            while (true) {
                sendData.poll()?.let {
                    Log.v(this@SignallingClient.javaClass.simpleName, "Sending: $it")
                    println("signalling client send: $it")
                    ws.send(it)
                }
            }
        } catch (exception: Throwable) {
            Log.e("asd","asd",exception)
        }
        
        client.dispatcher.executorService.shutdown()
    }


    fun send(dataObject: Any?) = runBlocking {
        sendChannel.send(gson.toJson(dataObject, dataObject!!::class.java))
    }

    fun destroy() {
        job.complete()
    }
}

CustomWebSocketListener.java (forgive me for the java + kotlin mixup, I am more comfortable with java)

// imports omitted
public class CustomWebSocketListener extends WebSocketListener {

    private final SignallingClientListener listener;
    
    public CustomWebSocketListener(SignallingClientListener listener) {
        super();
        this.listener = listener;
    }

    @Override
    public void onClosed(@NotNull WebSocket webSocket, int code, @NotNull String reason) {
        super.onClosed(webSocket, code, reason);
    }

    @Override
    public void onClosing(@NotNull WebSocket webSocket, int code, @NotNull String reason) {
        super.onClosing(webSocket, code, reason);
    }

    @Override
    public void onFailure(@NotNull WebSocket webSocket, @NotNull Throwable t, @Nullable Response response) {
        super.onFailure(webSocket, t, response);
        webSocket.close(1000, null);
    }

    @Override
    public void onMessage(@NotNull WebSocket webSocket, @NotNull String text) {
        super.onMessage(webSocket, text);
        JsonObject j = new Gson().fromJson(text, JsonObject.class);
        if (j.has("serverUrl")) {
            this.listener.onIceCandidateReceived(new Gson().fromJson(j, IceCandidate.class));
        } else if (j.has("type") && (j.get("type")).getAsString().equals("OFFER")) {
            this.listener.onOfferReceived(new Gson().fromJson(j, SessionDescription.class));
        } else if (j.has("type") && (j.get("type")).getAsString().equals("ANSWER")) {
            this.listener.onAnswerReceived(new Gson().fromJson(j, SessionDescription.class));
        }
    }

    @Override
    public void onMessage(@NotNull WebSocket webSocket, @NotNull ByteString bytes) {
        super.onMessage(webSocket, bytes);
    }

    @Override
    public void onOpen(@NotNull WebSocket webSocket, @NotNull Response response) {
        super.onOpen(webSocket, response);
    }
}

DataChannelObserver.java

// imports omitted
public class DataChannelObserver implements DataChannel.Observer {
    
    public DataChannelObserver(){}
    
    @Override
    public void onBufferedAmountChange(long l) {
        System.out.println("channel buffer amount changed");
    }

    @Override
    public void onStateChange() {
        System.out.println("channel state changed");
    }

    @Override
    public void onMessage(DataChannel.Buffer buffer) {
        System.out.println("channel msg: " + buffer.toString());
    }
}

and finally, the signalling server

'use strict';

const Uws = require('uWebSockets.js');
const uuid = require('uuid');

try {
    let listenSocket;
    
    let connectedClients = new Map();

    Uws.App({})
        .ws('/socket', {
            message: (ws, message, isBinary) => {
                const wsId = connectedClients.get(ws);
                for (const [socket, id] of connectedClients.entries()) {
                    if (socket !== ws && id !== wsId) {
                        console.log('sending message')
                        socket.send(message, isBinary);
                    }
                }
            },
            open: ws => {
                console.log('websocket opened');
                const wsId = uuid.v4();
                if (!connectedClients.has(ws)) {
                    connectedClients.set(ws, wsId);
                    console.log('added to connected clients: ' + wsId);
                } else {
                    console.log(wsId + ' already exists in connected clients')
                }
            },
        })
        .listen('0.0.0.0', Number.parseInt(process.env.PORT) || 9001, (token => {
            listenSocket = token;
            if (token) {
                console.log(`listening to port ${process.env.PORT || 9001}`);
            } else {
                console.log('failed to listen')
            }
        }))
} catch (e) {
    console.log(e);
    process.exit(1);
}

(I believe that the signalling server works, as it just forwards the requests to the other, and currently I am just testing with two devices. I modeled it off the example signalling server)

The SignallingClientListener and PeerConnectionObserver classes can be found in the example linked above, I did not modify them.

I tried creating the data channel before creating the offer, after creating it, after the connection has been established, and it is always null. I use the negotiated connection since I just want something really basic, and will be able to agree on the ID beforehand.

I've not set up my own STUN/TURN servers as my devices worked with the video calling example and so I imagine they should work with this, but let me know if I'm wrong.

Any help or tips about other possibly easier solutions would be greatly appreciated.

For reference, I am testing on two android 10 devices, and the webrtc version is 1.0.32006.

Answer 1

private fun buildPeerConnectionFactory(): PeerConnectionFactory {
        return PeerConnectionFactory
                .builder()
                .setOptions(PeerConnectionFactory.Options().apply {
                    disableEncryption = true // This is the cause of your null DataChannel and null IceCandidates
                    disableNetworkMonitor = true
                })
                .createPeerConnectionFactory()
    }

I have just figure it out it in about 10 minutes before after been stuck for almost a month, now finally I can have some new progress for my trash project

Answer 2

Answering my own question, not entirely but will post what worked for me. Ultimately I could not figure out what was wrong with the code above, I suspect somewhere I was not doing something right with the initialization or requests.

Some of the files are from above and have not been modified, but I'll post them anyways. Also, I had used this article as a starting point.

CustomPeerConnection.java

import android.content.Context;
import android.util.Log;

import org.webrtc.DataChannel;
import org.webrtc.IceCandidate;
import org.webrtc.MediaConstraints;
import org.webrtc.PeerConnection;
import org.webrtc.PeerConnectionFactory;
import org.webrtc.SessionDescription;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;


public class CustomPeerConnection {
    private final String TAG = this.getClass().getName();
    private PeerConnection peerConnection;
    private DataChannel dataChannel;
    private SignallingClient client;
    private CustomWebSocketListener wsListener;
    private Executor executor;
    
    public CustomPeerConnection(Context context, Executor executor) {
        this.executor = executor;
        init(context);
    }
    
    private void init(Context context) {
        PeerConnectionFactory.InitializationOptions initializationOptions = PeerConnectionFactory.InitializationOptions.builder(context)
                .createInitializationOptions();
        PeerConnectionFactory.initialize(initializationOptions);
        PeerConnectionFactory.Options options = new PeerConnectionFactory.Options();
        PeerConnectionFactory peerConnectionFactory = PeerConnectionFactory.builder().setOptions(options).createPeerConnectionFactory();
        List<PeerConnection.IceServer> iceServers = new ArrayList<>();
        iceServers.add(PeerConnection.IceServer.builder("stun:stun2.1.google.com:19302").createIceServer());
        iceServers.add(PeerConnection.IceServer.builder("stun:stun1.l.google.com:19302").createIceServer());
        iceServers.add(PeerConnection.IceServer.builder("turn:numb.viagenie.ca").setUsername("").setPassword("").createIceServer()); // go create your own turn server for free
        PeerConnection.RTCConfiguration rtcConfig = new PeerConnection.RTCConfiguration(iceServers);

        peerConnection = peerConnectionFactory.createPeerConnection(rtcConfig, new CustomPeerConnectionObserver() {

            @Override
            public void onIceCandidate(IceCandidate iceCandidate) {
                super.onIceCandidate(iceCandidate);
                client.send(iceCandidate);
                peerConnection.addIceCandidate(iceCandidate);
            }

            @Override
            public void onDataChannel(DataChannel dc) {
                super.onDataChannel(dc);
                executor.execute(() -> {
                    dataChannel = dc;
                    dataChannel.registerObserver(new CustomDataChannelObserver());
                    sendMessage(dataChannel, "test");
                });
            }

            @Override
            public void onIceConnectionChange(PeerConnection.IceConnectionState iceConnectionState) {
                super.onIceConnectionChange(iceConnectionState);
                if(iceConnectionState != null){
                    if(iceConnectionState == PeerConnection.IceConnectionState.CONNECTED){
                        Log.d(TAG, "IceConnectionState is CONNECTED");
                    }
                    if(iceConnectionState == PeerConnection.IceConnectionState.CLOSED){
                        Log.d(TAG, "IceConnectionState is CLOSED");
                    }
                    if(iceConnectionState == PeerConnection.IceConnectionState.FAILED){
                        Log.d(TAG, "IceConnectionState is FAILED");
                    }
                }
            }

            @Override
            public void onSignalingChange(PeerConnection.SignalingState signalingState) {
                super.onSignalingChange(signalingState);
                Log.d("Signalling State", signalingState.name());
                if (signalingState.equals(PeerConnection.SignalingState.HAVE_REMOTE_OFFER)) {
                    createAnswer();
                } 
            }
        });

        DataChannel.Init dcInit = new DataChannel.Init();
        dataChannel = peerConnection.createDataChannel("dataChannel", dcInit);
        dataChannel.registerObserver(new CustomDataChannelObserver());
        
        this.wsListener = new CustomWebSocketListener(this);
        this.client = new SignallingClient(wsListener);
        
    }
    
    public void sendMessage(DataChannel dataChannel, String msg) {
        ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
        boolean sent = dataChannel.send(new DataChannel.Buffer(buffer, false));
        Log.d("Message sent", "" + sent);
    }
    
    public void addIceCandidate(IceCandidate candidate) {
        this.peerConnection.addIceCandidate(candidate);
    }
    
    public void setRemoteDescription(CustomSdpObserver observer, SessionDescription sdp) {
        this.peerConnection.setRemoteDescription(observer, sdp);
    }

    public void createOffer(){
        MediaConstraints constraints = new MediaConstraints();
        constraints.optional.add(new MediaConstraints.KeyValuePair("DtlsSrtpKeyAgreement", "true"));
        constraints.optional.add(new MediaConstraints.KeyValuePair("internalSctpDataChannels", "true"));
        constraints.mandatory.add(new MediaConstraints.KeyValuePair("OfferToReceiveAudio", "false"));
        constraints.mandatory.add(new MediaConstraints.KeyValuePair("OfferToReceiveVideo", "false"));
//        constraints.mandatory.add(new MediaConstraints.KeyValuePair("IceRestart", "true"));
        
        peerConnection.createOffer(new CustomSdpObserver(){
            @Override
            public void onCreateSuccess(SessionDescription sessionDescription) {
                super.onCreateSuccess(sessionDescription);

                peerConnection.setLocalDescription(new CustomSdpObserver(), sessionDescription);
                client.send(sessionDescription);
            }

            @Override
            public void onCreateFailure(String s) {
                super.onCreateFailure(s);
                Log.d(TAG, "offer creation failed: " + s);
            }
        }, constraints);
    }
    
    public void createAnswer() {
        MediaConstraints constraints = new MediaConstraints();
        constraints.optional.add(new MediaConstraints.KeyValuePair("DtlsSrtpKeyAgreement", "true"));
        constraints.optional.add(new MediaConstraints.KeyValuePair("internalSctpDataChannels", "true"));
        constraints.mandatory.add(new MediaConstraints.KeyValuePair("OfferToReceiveAudio", "false"));
        constraints.mandatory.add(new MediaConstraints.KeyValuePair("OfferToReceiveVideo", "false"));
//        constraints.mandatory.add(new MediaConstraints.KeyValuePair("IceRestart", "true"));
        
        peerConnection.createAnswer(new CustomSdpObserver(){

            @Override
            public void onCreateSuccess(SessionDescription sessionDescription) {
                super.onCreateSuccess(sessionDescription);

                peerConnection.setLocalDescription(new CustomSdpObserver(), sessionDescription);
                client.send(sessionDescription);
            }

            @Override
            public void onCreateFailure(String s) {
                super.onCreateFailure(s);
                // Answer creation failed
                Log.d(TAG, "answer creation failed: " + s);
            }
        }, constraints);
    }

    public void close(){
        if(peerConnection == null){
            return;
        }

        if (dataChannel != null) {
            dataChannel.close();
        }
        peerConnection.close();
        peerConnection = null;
    }
    
    
}

CustomDataChannelObserver.java

import android.util.Log;

import org.webrtc.DataChannel;

import java.nio.ByteBuffer;

public class CustomDataChannelObserver implements DataChannel.Observer {
    public CustomDataChannelObserver(){
    }

    @Override
    public void onBufferedAmountChange(long l) {

    }

    @Override
    public void onStateChange() {
        Log.d("Channel Observer", "State has changed");
    }

    @Override
    public void onMessage(DataChannel.Buffer buffer) {
        ByteBuffer data = buffer.data;
        byte[] bytes = new byte[data.remaining()];
        data.get(bytes);
        final String message = new String(bytes);
        Log.d("Channel Observer", "msg: " + message);
    }
}

CustomPeerConnectionObserver.java

import android.util.Log;

import org.webrtc.DataChannel;
import org.webrtc.IceCandidate;
import org.webrtc.MediaStream;
import org.webrtc.PeerConnection;
import org.webrtc.RtpReceiver;

public class CustomPeerConnectionObserver implements PeerConnection.Observer {
    public CustomPeerConnectionObserver(){

    }

    @Override
    public void onSignalingChange(PeerConnection.SignalingState signalingState) {

    }

    @Override
    public void onIceConnectionChange(PeerConnection.IceConnectionState iceConnectionState) {

    }

    @Override
    public void onIceConnectionReceivingChange(boolean b) {

    }

    @Override
    public void onIceGatheringChange(PeerConnection.IceGatheringState iceGatheringState) {

    }

    @Override
    public void onIceCandidate(IceCandidate iceCandidate) {

    }

    @Override
    public void onIceCandidatesRemoved(IceCandidate[] iceCandidates) {

    }

    @Override
    public void onAddStream(MediaStream mediaStream) {

    }

    @Override
    public void onRemoveStream(MediaStream mediaStream) {

    }

    @Override
    public void onDataChannel(DataChannel dataChannel) {
        Log.d("CustomPeerConnectionObserver", dataChannel.label());
    }

    @Override
    public void onRenegotiationNeeded() {

    }

    @Override
    public void onAddTrack(RtpReceiver rtpReceiver, MediaStream[] mediaStreams) {

    }
}

CustomSdpObserver.java

import org.webrtc.SdpObserver;
import org.webrtc.SessionDescription;

public class CustomSdpObserver implements SdpObserver {

    public CustomSdpObserver(){

    }

    @Override
    public void onCreateSuccess(SessionDescription sessionDescription) {

    }

    @Override
    public void onSetSuccess() {

    }

    @Override
    public void onCreateFailure(String s) {

    }

    @Override
    public void onSetFailure(String s) {

    }
}

CustomWebSocketListener.java

import android.util.Log;

import com.google.gson.Gson;
import com.google.gson.JsonObject;

import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.webrtc.IceCandidate;
import org.webrtc.SessionDescription;

import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;
import okio.ByteString;

public class CustomWebSocketListener extends WebSocketListener {
    
    public CustomPeerConnection peerConnection;
    
    public CustomWebSocketListener(CustomPeerConnection peerConnection) {
        this.peerConnection = peerConnection;
    }
    
    @Override
    public void onClosed(@NotNull WebSocket webSocket, int code, @NotNull String reason) {
        super.onClosed(webSocket, code, reason);
//        System.out.println("Websocket closed");
    }

    @Override
    public void onClosing(@NotNull WebSocket webSocket, int code, @NotNull String reason) {
        super.onClosing(webSocket, code, reason);
        webSocket.close(1000, null);
    }

    @Override
    public void onFailure(@NotNull WebSocket webSocket, @NotNull Throwable t, @Nullable Response response) {
        super.onFailure(webSocket, t, response);
        System.out.println("websocket error: " + t.getMessage());
        webSocket.close(1000, null);
    }

    @Override
    public void onMessage(@NotNull WebSocket webSocket, @NotNull String text) {
        super.onMessage(webSocket, text);
        Log.d("socket listener", "received :" + text);
        JsonObject j = new Gson().fromJson(text, JsonObject.class);
//        Log.d("socket listener", "jsonobject keys: " + Arrays.deepToString(j.keySet().toArray()));
        if (j.has("serverUrl")) {
            peerConnection.addIceCandidate(new Gson().fromJson(j, IceCandidate.class));
        } else if (j.has("type") && (j.get("type")).getAsString().equals("OFFER")) {
            SessionDescription sdp = new SessionDescription(
                    SessionDescription.Type.valueOf(j.get("type").getAsString()),
                    j.get("description").getAsString());
            
            peerConnection.setRemoteDescription(new CustomSdpObserver(), sdp);
            
        } else if (j.has("type") && (j.get("type")).getAsString().equals("ANSWER")) {
            SessionDescription sdp = new SessionDescription(
                    SessionDescription.Type.valueOf(j.get("type").getAsString()),
                    j.get("description").getAsString());
            
            peerConnection.setRemoteDescription(new CustomSdpObserver(), sdp);
        }
    }

    @Override
    public void onMessage(@NotNull WebSocket webSocket, @NotNull ByteString bytes) {
        super.onMessage(webSocket, bytes);
//        System.out.println("bytes msg: " + bytes.hex());
    }

    @Override
    public void onOpen(@NotNull WebSocket webSocket, @NotNull Response response) {
        super.onOpen(webSocket, response);
//        System.out.println("Websocket opened");
        
    }
}

SignallingClient.kt

import android.util.Log
import com.google.gson.Gson
import io.ktor.util.KtorExperimentalAPI
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.ConflatedBroadcastChannel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import okhttp3.OkHttpClient
import okhttp3.Request
import okhttp3.WebSocket

@ExperimentalCoroutinesApi
@KtorExperimentalAPI
class SignallingClient(
    private val socketListener: CustomWebSocketListener
) : CoroutineScope {
    
    private val job = Job()
    private val gson = Gson()
    override val coroutineContext = Dispatchers.IO + job
    private val sendChannel = ConflatedBroadcastChannel<String>()
    
    private val url: String = "" // set your socket url here
    
    private lateinit var ws: WebSocket
    private lateinit var client: OkHttpClient

    init {
        connect()
    }
    
    private fun connect() = launch {
        client = OkHttpClient()
        val request = Request.Builder().url(url).build()
        ws = client.newWebSocket(request, socketListener)
        val sendData = sendChannel.openSubscription()
        try {
            while (true) {
                sendData.poll()?.let {
//                    Log.d(this@SignallingClient.javaClass.simpleName, "Sending: $it")
                    ws.send(it)
                }
            }
        } catch (exception: Throwable) {
            Log.e("asd","asd",exception)
        }
        
        client.dispatcher.executorService.shutdown()
    }


    fun send(dataObject: Any?) = runBlocking {
        sendChannel.send(gson.toJson(dataObject, dataObject!!::class.java))
    }

    fun destroy() {
        ws.close(1000, "destroy called");
        job.complete()
    }
}

WebRtc2Activity.java

import androidx.appcompat.app.AppCompatActivity;
import androidx.core.content.ContextCompat;

import android.os.Bundle;
import android.os.Handler;

import com.project.R;

import java.util.concurrent.Executor;

public class WebRtc2Activity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_web_rtc2);

        Executor g = ContextCompat.getMainExecutor(getApplicationContext());
        CustomPeerConnection peerConnection = new CustomPeerConnection(getApplicationContext(), g);

        Handler h = new Handler();
        h.postDelayed(() -> peerConnection.createOffer(), 3000);
//        peerConnection.createOffer();
    }
    
}

signalling server is the exact same as the one in the question.

Truth be told, the above code is decent starting code, and needs improvements. The way it's run in the activity is not done well, and there really isn't any error handling or dropped connection checks. I only did so to be able to make sure the mechanism works. So to anyone who may find this useful, do not run it the way I do here, modify it or extend.