Merge pull request #219 from angel-ayala/master

WebRTC streaming

app/build.gradle

@@ -8,7 +8,7 @@ android {
     compileSdkVersion 33
     defaultConfig {
         applicationId "sq.rogue.rosettadrone"
-        minSdkVersion 22
+        minSdkVersion 24
         targetSdkVersion 33
         versionCode 1
         versionName '5.2.0'    // This is the one to modify...
@@ -164,6 +164,17 @@ dependencies {
     implementation 'javax.mail:mail:1.4.7'                              // 1.4.7
     //  exclude 'javax.activation:activation:1.1'
     //  implementation 'com.sun.mail:android-activation:1.6.6'
+    // WebRTC library
+    implementation 'org.webrtc:google-webrtc:1.0.32006'
+    // Websocket implementation
+    implementation 'com.squareup.okhttp3:okhttp:4.10.0'
+    // Websocket utils for event-driven approach.
+    implementation 'io.reactivex.rxjava3:rxandroid:3.0.2'
+    // Because RxAndroid releases are few and far between, it is recommended you also
+    // explicitly depend on RxJava's latest version for bug fixes and new features.
+    // (see https://github.com/ReactiveX/RxJava/releases for latest 3.x.x version)
+    implementation 'io.reactivex.rxjava3:rxjava:3.1.5'
+
 }
 
 

app/src/main/AndroidManifest.xml

@@ -54,7 +54,9 @@
         android:label="@string/app_name"
         android:roundIcon="@mipmap/ic_launcher_round"
         android:supportsRtl="true"
-        android:theme="@style/AppTheme">
+        android:theme="@style/AppTheme"
+        android:usesCleartextTraffic="true"
+        tools:targetApi="m">
 
         <activity
             android:name="sq.rogue.rosettadrone.ConnectionActivity"
@@ -100,8 +102,8 @@
             tools:ignore="Instantiatable">
         </service>
 
-        <service android:name="sq.rogue.rosettadrone.video.VideoService">
-        </service>
+        <!--<service android:name="sq.rogue.rosettadrone.video.VideoService">
+        </service>-->
 
         <!-- DJI SDK needed also by emulator, this is a problem... -->
         <uses-library

app/src/main/java/sq/rogue/rosettadrone/MainActivity.java

@@ -376,6 +376,7 @@ public class MainActivity extends AppCompatActivity implements OnMapReadyCallbac
         */
 
         resumeVideo();
+        pluginManager.resume();
     }
 
     void resumeVideo() {
@@ -429,6 +430,7 @@ public class MainActivity extends AppCompatActivity implements OnMapReadyCallbac
     protected void onPause() {
         Log.e(TAG, "onPause()");
         super.onPause();
+        pluginManager.pause();
     }
 
     @Override

app/src/main/java/sq/rogue/rosettadrone/Plugin.java

@@ -3,13 +3,16 @@ package sq.rogue.rosettadrone;
 public class Plugin {
     protected PluginManager pluginManager;
 
-    protected void init(PluginManager pluginManager) {
+    public void init(PluginManager pluginManager) {
         this.pluginManager = pluginManager;
     }
 
     protected void start() {
     }
 
+    protected void pause() {}
+    protected void resume() {}
+
     /**
      * Video mode, resolution or codec changed.
      */
@@ -29,4 +32,13 @@ public class Plugin {
 
     public void settingsChanged() {
     }
+
+    public String getPrefString(String pref, String defPref) {
+        return pluginManager.mainActivity.sharedPreferences.getString(pref, defPref);
+    }
+
+    public boolean getPrefBoolean(String pref, boolean defPref) {
+        return pluginManager.mainActivity.sharedPreferences.getBoolean(pref, defPref);
+    }
+
 }

app/src/main/java/sq/rogue/rosettadrone/PluginManager.java

@@ -4,12 +4,14 @@
  */
 package sq.rogue.rosettadrone;
 
+import android.util.Log;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
 public class PluginManager {
-    private final String TAG = DroneModel.class.getSimpleName();
+    private final String TAG = PluginManager.class.getSimpleName();
 
     public MainActivity mainActivity;
 
@@ -21,7 +23,8 @@ public class PluginManager {
     //List<String> classNames = Arrays.asList("RawVideoStreamer", "AI9Tek");
     //List<String> classNames = Arrays.asList("AI9Tek");
 
-    List<String> classNames = Arrays.asList();
+    List<String> classNames = Arrays.asList("WebRTCStreaming");
+    //List<String> classNames = Arrays.asList();
 
     PluginManager(MainActivity mainActivity) {
         this.mainActivity = mainActivity;
@@ -36,7 +39,8 @@ public class PluginManager {
                 }
 
             } catch (Exception e) {
-                e.printStackTrace();
+//                e.printStackTrace();
+                Log.e(TAG, e.getClass().getName() + " occurred, stack trace: " + e.getMessage() + "\n" + e.getCause());
             }
         }
     }
@@ -53,6 +57,10 @@ public class PluginManager {
         }
     }
 
+    public void pause() { for (Plugin plugin : plugins) { plugin.pause(); } }
+
+    public void resume() { for (Plugin plugin : plugins) { plugin.resume(); } }
+
     public void onVideoChange() {
         for (Plugin plugin : plugins) {
             plugin.onVideoChange();

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/DJIStreamer.java

@@ -0,0 +1,82 @@
+package sq.rogue.rosettadrone.plugins.WebRTC;
+
+import android.content.Context;
+import android.util.Log;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.webrtc.VideoCapturer;
+import java.util.Hashtable;
+import dji.common.product.Model;
+import sq.rogue.rosettadrone.plugins.WebRTC.websocket.Socket;
+
+/**
+ * The DJIStreamer class will manage all ongoing P2P connections
+ * with clients, who desire videofeed.
+ */
+public class DJIStreamer {
+    private static final String TAG = DJIStreamer.class.getSimpleName();
+
+    private final Context context;
+    private final Hashtable<String, WebRTCClient> ongoingConnections = new Hashtable<>();
+    private final Model aircraftModel;
+
+    public DJIStreamer(Context context, Model aircraftModel, String stunServer){
+        this.aircraftModel = aircraftModel;
+        this.context = context;
+        setupSocketEvent(stunServer);
+    }
+
+    private WebRTCClient getClient(String socketID){
+        return ongoingConnections.getOrDefault(socketID, null);
+    }
+
+    private void removeClient(String socketID){
+        // TODO: Any other cleanup necessary?.. Let the client stop the VideoCapturer though.
+        ongoingConnections.remove(socketID);
+    }
+
+    private WebRTCClient addNewClient(String socketID, String stunServer){
+        VideoCapturer videoCapturer = new DJIVideoCapturer(aircraftModel);
+        WebRTCClient client = new WebRTCClient(socketID, context, videoCapturer, new WebRTCMediaOptions(), stunServer);
+        client.setConnectionChangedListener(new WebRTCClient.PeerConnectionChangedListener() {
+            @Override
+            public void onDisconnected() {
+                removeClient(client.peerSocketID);
+                Log.d(TAG, "DJIStreamer has removed connection from table. Remaining active sessions: " + ongoingConnections.size());
+            }
+        });
+        ongoingConnections.put(socketID, client);
+        return client;
+    }
+
+    private void setupSocketEvent(String stunServer){
+        Socket.getInstance().with(context).setOnEventResponseListener("webrtc_msg", (event, data) -> {
+            try {
+                JSONObject jsonData = new JSONObject(data);
+                String peerSocketID = jsonData.getString("socketID"); // The web-client sending a message
+
+                WebRTCClient client = getClient(peerSocketID);
+
+                if (client == null){
+                    // A new client wants to establish a P2P
+                    client = addNewClient(peerSocketID, stunServer);
+                }
+
+                // Then just pass the message to the client
+                client.handleWebRTCMessage(jsonData);
+            } catch (JSONException e) {
+                Log.e(TAG, "JSONException: " + e.getMessage());
+            }
+        });
+    }
+
+    public void closeVideoStream() {
+        Log.i(TAG, "closeVideoStream()");
+        ongoingConnections.keySet().forEach(socketID -> {
+            WebRTCClient client = getClient(socketID);
+            client.stopCapture();
+            client.close();
+            removeClient(socketID);
+        });
+    }
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/DJIVideoCapturer.java

@@ -0,0 +1,173 @@
+/* Originally sourced from
+* https://chromium.googlesource.com/external/webrtc/+/b6760f9e4442410f2bcb6090b3b89bf709e2fce2/webrtc/api/android/java/src/org/webrtc/CameraVideoCapturer.java
+* and rewritten to work for DJI drones.
+*  */
+package sq.rogue.rosettadrone.plugins.WebRTC;
+
+import org.webrtc.CapturerObserver;
+import org.webrtc.JavaI420Buffer;
+import org.webrtc.NV12Buffer;
+import org.webrtc.SurfaceTextureHelper;
+import org.webrtc.VideoCapturer;
+import org.webrtc.VideoFrame;
+
+import android.content.Context;
+import android.graphics.SurfaceTexture;
+import android.media.MediaCodecInfo;
+import android.media.MediaFormat;
+import android.os.SystemClock;
+import android.util.Log;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
+
+import dji.common.product.Model;
+import dji.sdk.camera.VideoFeeder;
+import dji.sdk.codec.DJICodecManager;
+
+public class DJIVideoCapturer implements VideoCapturer {
+    private final static String TAG = DJIVideoCapturer.class.getSimpleName();
+
+    private static DJICodecManager codecManager;
+    private static final ArrayList<CapturerObserver> observers = new ArrayList<CapturerObserver>();
+
+    private final Model aircraftModel;
+    private Context context;
+    private CapturerObserver capturerObserver;
+
+    public DJIVideoCapturer(Model aircraftModel){
+        this.aircraftModel = aircraftModel;
+    }
+
+    private void setupVideoListener(){
+        if(codecManager != null) {
+            Log.d(TAG, "codecManager not null");
+            return;
+        }
+
+        // Pass SurfaceTexture as null to force the Yuv callback - width and height for the surface texture does not matter
+        codecManager = new DJICodecManager(context, (SurfaceTexture)null, 0, 0);
+        codecManager.enabledYuvData(true);
+        codecManager.setYuvDataCallback(new DJICodecManager.YuvDataCallback() {
+            @Override
+            public void onYuvDataReceived(MediaFormat mediaFormat, ByteBuffer videoBuffer, int dataSize, int width, int height) {
+                if (videoBuffer != null){
+                    try{
+                        long timestampNS = TimeUnit.MILLISECONDS.toNanos(SystemClock.elapsedRealtime());
+                        VideoFrame videoFrame;
+                        // Check the color format. Could create more cases if needed
+                        int colorFormat = mediaFormat.getInteger(MediaFormat.KEY_COLOR_FORMAT);
+                        switch (colorFormat) {
+                            case MediaCodecInfo.CodecCapabilities.COLOR_FormatYUV420Flexible:
+                                // NV12 Buffer
+                                NV12Buffer nv12Buffer = new NV12Buffer(width, height,
+                                        mediaFormat.getInteger(MediaFormat.KEY_STRIDE),
+                                        mediaFormat.getInteger(MediaFormat.KEY_SLICE_HEIGHT),
+                                        videoBuffer, null);
+                                videoFrame = new VideoFrame(nv12Buffer, 0, timestampNS);
+                                break;
+                            case MediaCodecInfo.CodecCapabilities.COLOR_FormatYUV420Planar:
+                            case MediaCodecInfo.CodecCapabilities.COLOR_FormatYUV420SemiPlanar:
+                            case MediaCodecInfo.CodecCapabilities.COLOR_FormatYUV420PackedPlanar:
+                            case MediaCodecInfo.CodecCapabilities.COLOR_FormatYUV420PackedSemiPlanar:
+                                // I420 Buffer
+                                // Calculate the offsets and lengths for the Y, U, and V planes
+                                int ySize = width * height;
+                                int uvSize = ySize / 4;
+
+                                // Create YUV Buffers from the YUV data
+                                ByteBuffer yPlane = ByteBuffer.allocateDirect(ySize);
+                                ByteBuffer uPlane = ByteBuffer.allocateDirect(uvSize);
+                                ByteBuffer vPlane = ByteBuffer.allocateDirect(uvSize);
+
+                                // Copy data from videoBuffer to the respective planes
+                                videoBuffer.position(0);
+                                yPlane.put((ByteBuffer) videoBuffer.slice().limit(ySize));
+                                videoBuffer.position(ySize);
+                                uPlane.put((ByteBuffer) videoBuffer.slice().limit(uvSize));
+                                videoBuffer.position(ySize + uvSize);
+                                vPlane.put((ByteBuffer) videoBuffer.slice().limit(uvSize));
+
+                                yPlane.rewind(); uPlane.rewind(); vPlane.rewind();
+
+                                // Create JavaI420Buffer
+                                JavaI420Buffer i420Buffer = JavaI420Buffer.wrap(width, height,
+                                        yPlane, width,
+                                        uPlane, width / 2,
+                                        vPlane, width / 2, null);
+
+                                videoFrame = new VideoFrame(i420Buffer, 0, timestampNS);
+                                break;
+                            default:
+                                Log.e(TAG, "Color format: " + colorFormat + " is not implemented!!");
+                                return;
+                        }
+                        // Feed the video frame to everyone
+                        for (CapturerObserver obs : observers) {
+                            obs.onFrameCaptured(videoFrame);
+                        }
+                        videoFrame.release();
+                    } catch (Exception e){
+                        Log.e(TAG, e.getClass().getName() + " occurred, stack trace: " + e.getMessage() + "\n" + e.getCause());
+                    }
+                }
+            }
+        });
+
+        // Could create more cases if other drones from DJI require a different approach
+        if (aircraftModel.getDisplayName().equals("DJI Air 2S")){
+            // The Air 2S relies on the VideoDataListener to obtain the video feed
+            // The onReceive callback provides us the raw H264 (at least according to official documentation). To decode it we send it to our DJICodecManager
+            // H264 or H265 encoding is done to compress and save bandwidth. (4K video might force a switch to H265 on DJI drones)
+            VideoFeeder.VideoDataListener videoDataListener = new VideoFeeder.VideoDataListener() {
+                @Override
+                public void onReceive(byte[] bytes, int dataSize) {
+                    // Pass the encoded data along to obtain the YUV-color data
+                    codecManager.sendDataToDecoder(bytes, dataSize);
+                }
+            };
+            VideoFeeder.getInstance().getPrimaryVideoFeed().addVideoDataListener(videoDataListener);
+        }
+    }
+
+    @Override
+    public void initialize(SurfaceTextureHelper surfaceTextureHelper, Context applicationContext,
+                           CapturerObserver capturerObserver) {
+        this.context = applicationContext;
+        this.capturerObserver = capturerObserver;
+
+        observers.add(capturerObserver);
+    }
+
+    @Override
+    public void startCapture(int width, int height, int framerate) {
+        // Hook onto the DJI onYuvDataReceived event
+        setupVideoListener();
+    }
+
+    @Override
+    public void stopCapture() throws InterruptedException {
+        codecManager.enabledYuvData(false);
+        codecManager.setYuvDataCallback(null);
+        codecManager.destroyCodec();
+        codecManager = null;
+    }
+
+    @Override
+    public void changeCaptureFormat(int width, int height, int framerate) {
+        // Empty on purpose
+    }
+
+    @Override
+    public void dispose() {
+        // Stop receiving frames on the callback from the decoder
+        if (observers.contains(capturerObserver))
+            observers.remove(capturerObserver);
+    }
+
+    @Override
+    public boolean isScreencast() {
+        return false;
+    }
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/SimpleSdpObserver.java

@@ -0,0 +1,24 @@
+package sq.rogue.rosettadrone.plugins.WebRTC;
+
+import org.webrtc.SdpObserver;
+import org.webrtc.SessionDescription;
+
+class SimpleSdpObserver implements SdpObserver {
+
+    @Override
+    public void onCreateSuccess(SessionDescription sessionDescription) {
+    }
+
+    @Override
+    public void onSetSuccess() {
+    }
+
+    @Override
+    public void onCreateFailure(String s) {
+    }
+
+    @Override
+    public void onSetFailure(String s) {
+    }
+
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/WebRTCClient.java

@@ -0,0 +1,273 @@
+package sq.rogue.rosettadrone.plugins.WebRTC;
+
+import static org.webrtc.SessionDescription.Type.OFFER;
+
+import android.content.Context;
+import android.util.Log;
+import java.util.ArrayList;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.webrtc.DataChannel;
+import org.webrtc.DefaultVideoDecoderFactory;
+import org.webrtc.DefaultVideoEncoderFactory;
+import org.webrtc.EglBase;
+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.SessionDescription;
+import org.webrtc.VideoCapturer;
+import org.webrtc.VideoSource;
+import org.webrtc.VideoTrack;
+
+import sq.rogue.rosettadrone.plugins.WebRTC.websocket.Socket;
+
+public class WebRTCClient {
+    private static final String TAG = WebRTCClient.class.getSimpleName();
+    private final Context context;
+
+    // WebRTC related variables
+    private PeerConnection peerConnection;
+    private VideoTrack videoTrackFromCamera;
+    private final WebRTCMediaOptions options;
+    private final VideoCapturer videoCapturer;
+
+    private PeerConnectionChangedListener connectionChangedListener;
+    public void setConnectionChangedListener(PeerConnectionChangedListener connectionChangedListener) { this.connectionChangedListener = connectionChangedListener; }
+
+    // Peer variables of the client requesting a stream
+    public final String peerSocketID;
+
+    private static PeerConnectionFactory factory;
+    private static PeerConnectionFactory getFactory(Context context){
+        if (factory == null) {
+            initializeFactory(context);
+        }
+        return factory;
+    }
+
+    public WebRTCClient(String peerSocketID, Context context, VideoCapturer videoCapturer, WebRTCMediaOptions options, String stunServer){
+        this.peerSocketID = peerSocketID;
+        this.context = context;
+        this.options = options;
+        this.videoCapturer = videoCapturer;
+
+        createVideoTrackFromVideoCapturer();
+        initializePeerConnection(stunServer);
+        startStreamingVideo();
+    }
+
+    private static void initializeFactory(Context context){
+        // EglBase seems to be used for Hardware-acceleration for our video. Could help with smoothing things. (keep it)
+        EglBase rootEglBase = EglBase.create();
+
+        // Initialize the PeerConnectionFactory
+        PeerConnectionFactory.InitializationOptions options = PeerConnectionFactory.InitializationOptions.builder(context)
+                .setEnableInternalTracer(true)
+                .setFieldTrials("WebRTC-H264HighProfile/Enabled/")
+                .createInitializationOptions();
+        PeerConnectionFactory.initialize(options);
+
+        // Now configure and build the factory
+        factory = PeerConnectionFactory
+                .builder()
+                .setVideoDecoderFactory(new DefaultVideoDecoderFactory(rootEglBase.getEglBaseContext()))
+                .setVideoEncoderFactory(new DefaultVideoEncoderFactory(rootEglBase.getEglBaseContext(), true, true))
+                .setOptions(new PeerConnectionFactory.Options()).createPeerConnectionFactory();
+    }
+
+    public void handleWebRTCMessage(JSONObject message){
+        try {
+            // Log.d(TAG, "connectToSignallingServer: got message " + message);
+            if (message.getString("type").equals("offer")) {
+                Log.d(TAG, "connectToSignallingServer: received an offer");
+                peerConnection.setRemoteDescription(new SimpleSdpObserver(), new SessionDescription(OFFER, message.getString("sdp")));
+                answerCall();
+            } else if (message.getString("type").equals("candidate")) {
+                Log.d(TAG, "connectToSignallingServer: receiving candidates");
+                IceCandidate candidate = new IceCandidate(message.getString("id"), message.getInt("label"), message.getString("candidate"));
+                peerConnection.addIceCandidate(candidate);
+            }
+        }
+        catch (JSONException e) {
+            Log.e(TAG, "JSONException: " + e.getMessage());
+        }
+    }
+
+    private void answerCall() {
+        peerConnection.createAnswer(new SimpleSdpObserver() {
+            @Override
+            public void onCreateSuccess(SessionDescription sessionDescription) {
+                peerConnection.setLocalDescription(new SimpleSdpObserver(), sessionDescription);
+                JSONObject message = new JSONObject();
+                try {
+                    message.put("type", "answer");
+                    message.put("sdp", sessionDescription.description);
+                    sendMessage(message);
+                } catch (JSONException e) {
+                    Log.e(TAG, "JSONException: " + e.getMessage());
+                }
+            }
+        }, new MediaConstraints());
+    }
+
+    private void sendMessage(Object message) {
+        // Log.d(TAG, "Emitting message to " + peerSocketID);
+        try {
+            // append target socketID to sent data.
+            JSONObject sendMessage = new JSONObject(message.toString());
+            sendMessage.put("socketID", peerSocketID);
+
+            Socket.getInstance().send("webrtc_msg", sendMessage);
+        } catch (JSONException e) {
+            Log.e(TAG, "JSONException: " + e.getMessage());
+        }
+    }
+
+    private void createVideoTrackFromVideoCapturer() {
+        VideoSource videoSource = getFactory(context).createVideoSource(false);
+
+        // Instantiate our custom video capturer to get video from our drone
+        videoCapturer.initialize(null, context, videoSource.getCapturerObserver());
+        videoCapturer.startCapture(options.VIDEO_RESOLUTION_WIDTH, options.VIDEO_RESOLUTION_HEIGHT, options.FPS);
+
+        videoTrackFromCamera = getFactory(context).createVideoTrack(options.VIDEO_SOURCE_ID, videoSource);
+        videoTrackFromCamera.setEnabled(true);
+    }
+
+    private void initializePeerConnection(String stunServer) {
+        peerConnection = createPeerConnection(stunServer);
+    }
+
+    private void startStreamingVideo() {
+        MediaStream mediaStream = getFactory(context).createLocalMediaStream(options.MEDIA_STREAM_ID);
+        mediaStream.addTrack(videoTrackFromCamera);
+        peerConnection.addStream(mediaStream);
+    }
+
+    public void stopCapture() {
+        // Stop video capture if it is active
+        try {
+            if (videoCapturer != null) {
+                videoCapturer.stopCapture();
+            }
+        } catch (InterruptedException e) {
+            Log.e(TAG, "Error stopping video capture: " + e.getMessage());
+        } finally {
+            // Dispose of video capturer to free its resources
+            if (videoCapturer != null) {
+                videoCapturer.dispose();
+                Log.d(TAG, "Video capturer disposed.");
+            }
+            Log.d(TAG, "WebRTC client resources clean complete.");
+        }
+    }
+
+    public void close() {
+        // Close the peer connection if it exists
+        if (peerConnection != null) {
+            peerConnection.close();
+            peerConnection.dispose();
+            Log.d(TAG, "PeerConnection disposed.");
+        }
+
+        // Optionally dispose of the PeerConnectionFactory if this is the last WebRTCClient instance
+        if (factory != null) {
+            factory.dispose();
+            factory = null;
+            Log.d(TAG, "PeerConnectionFactory disposed.");
+        }
+
+        Log.d(TAG, "WebRTC client close complete.");
+    }
+
+    private PeerConnection createPeerConnection(String stunServer) {
+        ArrayList<PeerConnection.IceServer> iceServers = new ArrayList<>();
+        PeerConnection.IceServer stun =  PeerConnection.IceServer.builder(stunServer).createIceServer();
+        iceServers.add(stun);
+        PeerConnection.RTCConfiguration rtcConfig = new PeerConnection.RTCConfiguration(iceServers);
+
+        PeerConnection.Observer pcObserver = new PeerConnection.Observer() {
+            @Override
+            public void onSignalingChange(PeerConnection.SignalingState signalingState) {
+                Log.d(TAG, "onSignalingChange: ");
+            }
+
+            @Override
+            public void onIceConnectionChange(PeerConnection.IceConnectionState iceConnectionState) {
+                switch (iceConnectionState){
+                    case DISCONNECTED:
+                        Log.d(TAG, "PEER HAS DISCONNECTED");
+                        if (connectionChangedListener != null)
+                            connectionChangedListener.onDisconnected();
+                        // Dispose of the capturer and then the peer connection to clean up properly
+                        videoCapturer.dispose();
+                        break;
+                }
+            }
+
+            @Override
+            public void onIceConnectionReceivingChange(boolean b) {
+                Log.d(TAG, "onIceConnectionReceivingChange: ");
+            }
+
+            @Override
+            public void onIceGatheringChange(PeerConnection.IceGatheringState iceGatheringState) {
+                Log.d(TAG, "onIceGatheringChange: ");
+            }
+
+            @Override
+            public void onAddTrack(RtpReceiver rtpReceiver, MediaStream[] mediaStreams) {
+                // We are not interested in displaying whatever video feed we receive from the other end.
+                // Not that we are getting any..
+            }
+
+            @Override
+            public void onIceCandidate(IceCandidate iceCandidate) {
+                Log.d(TAG, "onIceCandidate: ");
+                JSONObject message = new JSONObject();
+
+                try {
+                    message.put("type", "candidate");
+                    message.put("label", iceCandidate.sdpMLineIndex);
+                    message.put("id", iceCandidate.sdpMid);
+                    message.put("candidate", iceCandidate.sdp);
+
+                    sendMessage(message);
+                } catch (JSONException e) {
+                    Log.e(TAG, "JSONException " + e.getMessage());
+                }
+            }
+
+            @Override
+            public void onIceCandidatesRemoved(IceCandidate[] iceCandidates) { Log.d(TAG, "onIceCandidatesRemoved: "); }
+
+            @Override
+            public void onAddStream(MediaStream mediaStream) { Log.d(TAG, "onAddStream: "); }
+
+            @Override
+            public void onRemoveStream(MediaStream mediaStream) {
+                Log.d(TAG, "onRemoveStream: ");
+            }
+
+            @Override
+            public void onDataChannel(DataChannel dataChannel) {
+                Log.d(TAG, "onDataChannel: ");
+            }
+
+            @Override
+            public void onRenegotiationNeeded() {
+                Log.d(TAG, "onRenegotiationNeeded: ");
+            }
+        };
+
+        return getFactory(context).createPeerConnection(rtcConfig, pcObserver);
+    }
+
+    public interface PeerConnectionChangedListener {
+        void onDisconnected(); // Is called when our peer disconnects from the call
+    }
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/WebRTCMediaOptions.java

@@ -0,0 +1,9 @@
+package sq.rogue.rosettadrone.plugins.WebRTC;
+
+public class WebRTCMediaOptions {
+    public String MEDIA_STREAM_ID = "Phantom4Pro";
+    String VIDEO_SOURCE_ID = "Phantom4Prov0";
+    int VIDEO_RESOLUTION_WIDTH = 320;
+    int VIDEO_RESOLUTION_HEIGHT = 240;
+    int FPS = 30;
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/OnEventListener.java

@@ -0,0 +1,10 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+public interface OnEventListener {
+    /**
+     * Invoked when new message received from websocket with {event, data} structure
+     *
+     * @param data Data string received
+     */
+    void onMessage(String data);
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/OnEventResponseListener.java

@@ -0,0 +1,11 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+public interface OnEventResponseListener {
+    /**
+     * Invoked when new message received from websocket with {event, data} structure
+     *
+     * @param event message event
+     * @param data data string received
+     */
+    void onMessage(String event, String data);
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/OnMessageListener.java

@@ -0,0 +1,10 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+public interface OnMessageListener {
+    /**
+     * Invoked when new message received from websocket
+     *
+     * @param data Data string received
+     */
+    void onMessage(String data);
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/OnStateChangeListener.java

@@ -0,0 +1,40 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+import okhttp3.Response;
+
+public abstract class OnStateChangeListener {
+    /**
+     * Invoked when a web socket has been accepted by the remote peer and may begin transmitting
+     * messages.
+     */
+    public void onOpen(Response response) {
+    }
+
+    /**
+     * Invoked when both peers have indicated that no more messages will be transmitted and the
+     * connection has been successfully released. No further calls to this listener will be made.
+     */
+    public void onClosed(int code, String reason) {
+    }
+
+    /**
+     * Invoked when a web socket has been closed due to an error reading from or writing to the
+     * network. Both outgoing and incoming messages may have been lost.
+     */
+    public void onFailure(Throwable t) {
+    }
+
+    /**
+     * Invoked when a web socket has been closed due to an error and reconnection attempt is started.
+     */
+    public void onReconnect(int attemptsCount, long attemptDelay) {
+    }
+
+    /**
+     * Invoked when new socket connection status changed.
+     *
+     * @param status new socket status
+     */
+    public void onChange(SocketState status) {
+    }
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/Socket.java

@@ -0,0 +1,477 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+/*
+ * For logging I use `com.orhanobut:logger` Logger
+ */
+
+import android.os.Handler;
+import android.os.Looper;
+import androidx.annotation.NonNull;
+import android.util.Log;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import java.net.ProtocolException;
+import java.util.HashMap;
+import java.util.Map;
+
+import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
+import io.reactivex.rxjava3.disposables.CompositeDisposable;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.functions.Consumer;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.subjects.PublishSubject;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.WebSocket;
+import okhttp3.WebSocketListener;
+import okhttp3.internal.ws.RealWebSocket;
+import okio.ByteString;
+
+/**
+ * Websocket class based on OkHttp3 with {event->data} message format to make your life easier.
+ *
+ * @author Ali Yusuf
+ */
+
+public class Socket {
+
+    private final static String TAG = Socket.class.getSimpleName();
+    private final static String CLOSE_REASON = "End of session";
+    private final static int MAX_COLLISION = 7;
+
+    private static Socket mInstance = null;
+
+    /**
+     * Websocket state
+     */
+    private SocketState mState;
+    /**
+     * Websocket main request
+     */
+    private Request mRequest;
+    /**
+     * Websocket http client
+     */
+    private OkHttpClient.Builder mHttpClient;
+    /**
+     * Websocket connection
+     */
+    private RealWebSocket mRealWebSocket;
+    /**
+     * Reconnection post delayed handler
+     */
+    private Handler mHandler;
+    /**
+     * Stores number of reconnecting attempts
+     */
+    private int reconnectionAttempts;
+    /**
+     * Indicate if it's termination to stop reconnecting
+     */
+    private boolean isForceTermination;
+    /**
+     * Socket event bus
+     */
+    private PublishSubject<Object> eventBus = PublishSubject.create();
+    /**
+     * Map that's help to keep track with hole lifecycle,
+     * used to cancel all lifecycle subscriptions.
+     *
+     * lifecycle -> [events] map
+     */
+    private Map<Object, CompositeDisposable> sSubscriptionsMap = new HashMap<>();
+    /**
+     * Map that's help to keep track with lifecycle subscriptions with corresponding
+     * event and listener. Used to cancel particular subscription or reset it.
+     *
+     * lifecycle -> {event -> {listener -> subscription}} map
+     */
+    private Map<Object, Map<Class, Map<Object, Disposable>>> sListenerBinderMap = new HashMap<>();
+
+    Socket() {}
+
+    @NonNull
+    public static Socket getInstance() {
+        if(mInstance == null) {
+            throw new AssertionError("Make sure to use SocketBuilder before using Socket#getInstance.");
+        }
+        return mInstance;
+    }
+
+    static Socket init(OkHttpClient.Builder httpClient, Request request) {
+        mInstance = new Socket();
+        mInstance.mHttpClient = httpClient;
+        mInstance.mRequest = request;
+        mInstance.mState = SocketState.CLOSED;
+        mInstance.mHandler = new Handler(Looper.getMainLooper());
+        mInstance.isForceTermination = false;
+        return mInstance;
+    }
+
+    /**
+     * Start socket connection if it's not already started
+     */
+    public void connect() {
+        if (mInstance.mHttpClient == null || mInstance.mRequest == null) {
+            throw new IllegalStateException("Make sure to use SocketBuilder before using Socket#connect.");
+        }
+        if (mRealWebSocket == null || mState == SocketState.CLOSED) {
+            mRealWebSocket = (RealWebSocket) mHttpClient.build().newWebSocket(mRequest, webSocketListener);
+            changeState(SocketState.OPENING);
+        }
+    }
+
+    /**
+     * Send message in {event->data} format
+     *
+     * @param event event name that you want sent message to
+     * @param data message data object
+     * @return true if the message send/on socket send quest; false otherwise
+     */
+    public boolean send(@NonNull String event, @NonNull Object data){
+        return send(event, data.toString());
+    }
+
+    /**
+     * Send message in {event->data} format
+     *
+     * @param event event name that you want sent message to
+     * @param data message data in JSON format
+     * @return true if the message send/on socket send quest; false otherwise
+     */
+    public boolean send(@NonNull String event, @NonNull String data){
+        if (mRealWebSocket != null && mState == SocketState.OPEN) {
+            try {
+                JSONObject text = new JSONObject();
+                text.put("event", event);
+                text.put("data", new JSONObject(data));
+
+                return mRealWebSocket.send(text.toString());
+            } catch (JSONException e) {
+                Log.e(TAG, "JSONException when sending data: " + e.getMessage());
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Set global listener which fired every time message received with contained data.
+     *
+     * @param listener message on arrive listener
+     */
+    public Socket addOnEventListener(@NonNull String event, @NonNull OnEventListener listener){
+        with(this).addOnEventListener(event, listener);
+        return this;
+    }
+
+    /**
+     * Set global listener which fired every time message received with contained data.
+     *
+     * @param listener message on arrive listener
+     */
+    public Socket addOnEventResponseListener(@NonNull String event, @NonNull OnEventResponseListener listener){
+        with(this).setOnEventResponseListener(event, listener);
+        return this;
+    }
+
+    /**
+     * Set global state listener which fired every time {@link Socket#mState} changed.
+     *
+     * @param listener state change listener
+     */
+    public Socket addOnChangeStateListener(@NonNull OnStateChangeListener listener) {
+        with(this).addOnChangeStateListener(listener);
+        return this;
+    }
+
+    /**
+     * Set global message listener which will be called in any message received even if it's not
+     * in a {event -> data} format.
+     *
+     * @param listener message listener
+     */
+    public Socket addMessageListener(@NonNull OnMessageListener listener) {
+        with(this).addOnMessageListener(listener);
+        return this;
+    }
+
+    /**
+     * Send normal close request to the host
+     */
+    public void close() {
+        if (mRealWebSocket != null) {
+            mRealWebSocket.close(1000, CLOSE_REASON);
+        }
+    }
+
+    /**
+     * Send close request to the host
+     */
+    public void close(int code, @NonNull String reason) {
+        if (mRealWebSocket != null) {
+            mRealWebSocket.close(code, reason);
+        }
+    }
+
+    /**
+     * Terminate the socket connection permanently
+     */
+    public void terminate() {
+        isForceTermination = true; // skip onFailure callback
+        if (mRealWebSocket != null) {
+            mRealWebSocket.cancel(); // close connection
+            mRealWebSocket = null; // clear socket object
+        }
+        changeState(SocketState.CLOSED);
+        postEvent(new SocketEvents.CloseStatusEvent(1006, ""));
+    }
+
+    /**
+     * Retrieve current socket connection state {@link SocketState}
+     */
+    public SocketState getState() {
+        return mState;
+    }
+
+    /**
+     * Change current state and call listener method with new state
+     * {@link OnStateChangeListener#onChange(SocketState)}
+     * @param newState new state
+     */
+    private void changeState(SocketState newState) {
+        mState = newState;
+        postEvent(new SocketEvents.ChangeStatusEvent(newState));
+    }
+
+    /**
+     * Try to reconnect to the websocket after delay time using <i>Exponential backoff</i> method.
+     * @see <a href="https://en.wikipedia.org/wiki/Exponential_backoff"></a>
+     */
+    private void reconnect() {
+        if (mState != SocketState.CONNECT_ERROR) // connection not closed !!
+            return;
+
+        changeState(SocketState.RECONNECT_ATTEMPT);
+
+        if (mRealWebSocket != null) {
+            // Cancel websocket connection
+            mRealWebSocket.cancel();
+            // Clear websocket object
+            mRealWebSocket = null;
+        }
+
+        // Calculate delay time
+        int collision = reconnectionAttempts > MAX_COLLISION ? MAX_COLLISION : reconnectionAttempts;
+        long delayTime = Math.round((Math.pow(2, collision)-1)/2) * 1000;
+
+
+        postEvent(new SocketEvents.ReconnectStatusEvent(reconnectionAttempts + 1, delayTime));
+
+        // Remove any pending posts of callbacks
+        mHandler.removeCallbacksAndMessages(null);
+        // Start new post delay
+        mHandler.postDelayed(() -> {
+            changeState(SocketState.RECONNECTING);
+            reconnectionAttempts++; // Increment connections attempts
+            connect(); // Establish new connection
+        }, delayTime);
+    }
+
+    private WebSocketListener webSocketListener = new WebSocketListener() {
+        @Override
+        public void onOpen(WebSocket webSocket, Response response) {
+            Log.d(TAG, "Socket has been opened successfully.");
+            // reset connections attempts counter
+            reconnectionAttempts = 0;
+
+            // fire open event listener
+            changeState(SocketState.OPEN);
+            postEvent(new SocketEvents.OpenStatusEvent(response));
+        }
+
+        /**
+         * Accept only Json data with format:
+         * <b> {"event":"event name","data":{some data ...}} </b>
+         */
+        @Override
+        public void onMessage(WebSocket webSocket, String text) {
+            // print received message in log
+            // Log.d(TAG, "New Message received \n" + text);
+
+            // call message listener
+            postEvent(new SocketEvents.BaseMessageEvent(text));
+
+            try {
+                // Parse message text
+                JSONObject response = new JSONObject(text);
+                String event = response.getString("event");
+                JSONObject data = response.getJSONObject("data");
+
+                // call event listener with received data
+                postEvent(new SocketEvents.ResponseMessageEvent(event, data.toString()));
+
+                // call event listener
+                postEvent(new SocketEvents.BaseMessageEvent(event));
+            } catch (JSONException e) {
+                // Message text not in JSON format or don't have {event}|{data} object
+                Log.e(TAG,"JSONException:" + e.getMessage());
+            }
+        }
+
+        @Override
+        public void onMessage(WebSocket webSocket, ByteString bytes) {
+            // TODO: some action
+        }
+
+        @Override
+        public void onClosing(WebSocket webSocket, int code, String reason) {
+            Log.d(TAG, "Close request from server with reason: " + reason);
+            changeState(SocketState.CLOSING);
+            webSocket.close(1000,reason);
+        }
+
+        @Override
+        public void onClosed(WebSocket webSocket, int code, String reason) {
+            Log.d(TAG, "Close request from server with reason: " + reason);
+            changeState(SocketState.CLOSED);
+            postEvent(new SocketEvents.CloseStatusEvent(code, reason));
+        }
+
+        /**
+         * This method call if:
+         * - Fail to verify websocket GET request  => Throwable {@link ProtocolException}
+         * - Can't establish websocket connection after upgrade GET request => response null, Throwable {@link Exception}
+         * - First GET request had been failed => response null, Throwable {@link java.io.IOException}
+         * - Fail to send Ping => response null, Throwable {@link java.io.IOException}
+         * - Fail to send data frame => response null, Throwable {@link java.io.IOException}
+         * - Fail to read data frame => response null, Throwable {@link java.io.IOException}
+         */
+        @Override
+        public void onFailure(WebSocket webSocket, Throwable t, Response response) {
+            if (!isForceTermination) {
+                isForceTermination = false; // reset flag
+                Log.e(TAG, "Socket connection fail, try to reconnect. (" + reconnectionAttempts + ")", t);
+                changeState(SocketState.CONNECT_ERROR);
+                reconnect();
+            }
+            postEvent(new SocketEvents.FailureStatusEvent(t));
+        }
+    };
+
+    /**
+     * State subscription this lifecycle to socket events and listen for updates on that event.
+     *
+     * Note: Make sure to call {@link Socket#unsubscribe(Object)} to avoid memory leaks.
+     */
+    public SocketListenersBinder with(Object lifecycle){
+        return new SocketListenersBinder(lifecycle, this);
+    }
+
+    /**
+     * Post an event for all subscribers of that event.
+     */
+    private void postEvent(@NonNull Object event) {
+        if (eventBus.hasObservers()){
+            eventBus.onNext(event);
+        }
+    }
+
+    /**
+     * Unregisters this object from the listeners bus, removing all subscriptions.
+     * This should be called when the object is going to go out of memory.
+     */
+    public void unsubscribe(Object lifecycle){
+        CompositeDisposable compositeSubscription = sSubscriptionsMap.remove(lifecycle);
+        if (compositeSubscription != null) {
+            compositeSubscription.dispose();
+            // clear lifecycle subscriptions of event
+            sListenerBinderMap.remove(lifecycle);
+        }
+    }
+
+    /**
+     * Get the CompositeDisposable or create it if it's not already in memory.
+     */
+    @NonNull
+    private CompositeDisposable getCompositeSubscription(@NonNull Object object) {
+        CompositeDisposable compositeSubscription = sSubscriptionsMap.get(object);
+        if (compositeSubscription == null) {
+            compositeSubscription = new CompositeDisposable();
+            sSubscriptionsMap.put(object, compositeSubscription);
+        }
+        return compositeSubscription;
+    }
+
+    /**
+     * Get the event -> disposable map of the specific lifecycle.
+     */
+    private Map<Class, Map<Object, Disposable>> getListenerBinderMap(Object lifecycle) {
+        Map<Class, Map<Object, Disposable>> disposableMap = sListenerBinderMap.get(lifecycle);
+        if (disposableMap == null) {
+            disposableMap = new HashMap<>();
+            sListenerBinderMap.put(lifecycle, disposableMap);
+        }
+        return disposableMap;
+    }
+
+    /**
+     * Add event subscription to the specified lifecycle and listen for updates on that event,
+     * each listener subscription must be unique one each lifecycle, event.
+     * Old subscription of same listener {@code listener} will be disposed if exist.
+     */
+    <T> void addEventSubscription(Object lifecycle, Class<T> eventClass, @NonNull Consumer<T> consumer, Object listener) {
+        Map<Object, Disposable> disposableMap = getListenerBinderMap(lifecycle).get(eventClass);
+        if (disposableMap == null) {
+            disposableMap = new HashMap<>();
+        } else {
+            // remove old subscription if exist
+            removeEventSubscriptionListener(lifecycle, eventClass, listener);
+        }
+
+        // add event subscription to the bus event
+        Disposable disposable = eventBus.filter(o -> (o != null)) // Filter out null objects, better safe than sorry
+                .filter(eventClass::isInstance)
+                .cast(eventClass) // Cast it for easier usage
+                .subscribeOn(Schedulers.io())
+                .observeOn(AndroidSchedulers.mainThread())
+                .subscribe(consumer);
+        getCompositeSubscription(lifecycle).add(disposable);
+
+        // update lifecycle subscriptions
+        disposableMap.put(listener, disposable);
+        getListenerBinderMap(lifecycle).put(eventClass, disposableMap);
+    }
+
+    /**
+     * Remove all event subscriptions of the specified lifecycle.
+     */
+    <T> void removeEventSubscriptions(Object lifecycle, Class<T> eventClass) {
+        // clear lifecycle subscriptions of event
+        Map<Object, Disposable> disposableMap = sListenerBinderMap.get(lifecycle).remove(eventClass);
+        if (disposableMap != null) {
+            for (Disposable disposable : disposableMap.values()) {
+                if (disposable != null) {
+                    getCompositeSubscription(lifecycle).remove(disposable); // stop subscription
+                }
+            }
+        }
+    }
+
+    /**
+     * Remove event listener subscription of the specified lifecycle.
+     */
+    <T> void removeEventSubscriptionListener(Object lifecycle, Class<T> eventClass, Object listener) {
+        // get subscription
+        Map<Object, Disposable> disposableMap = sListenerBinderMap.get(lifecycle).get(eventClass);
+        if (disposableMap != null) {
+            Disposable disposable = disposableMap.remove(listener);
+            if (disposable != null) {
+                getCompositeSubscription(lifecycle).remove(disposable); // stop subscription
+            }
+        }
+    }
+
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/SocketBuilder.java

@@ -0,0 +1,42 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+import androidx.annotation.NonNull;
+import java.util.concurrent.TimeUnit;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+
+/**
+ * Builder class to build websocket connection
+ */
+public class SocketBuilder {
+
+    private Request.Builder request;
+
+    private OkHttpClient.Builder httpClient = new OkHttpClient.Builder();
+
+    private SocketBuilder(Request.Builder request) {
+        this.request = request;
+    }
+
+    public static SocketBuilder with(@NonNull String url) {
+        // Silently replace web socket URLs with HTTP URLs.
+        if (!url.regionMatches(true, 0, "ws:", 0, 3) && !url.regionMatches(true, 0, "wss:", 0, 4))
+            throw new IllegalArgumentException("web socket url must start with ws or wss, passed url is " + url);
+
+        return new SocketBuilder(new Request.Builder().url(url));
+    }
+
+    public SocketBuilder setPingInterval(long interval, @NonNull TimeUnit unit){
+        httpClient.pingInterval(interval, unit);
+        return this;
+    }
+
+    public SocketBuilder addHeader(@NonNull String name, @NonNull String value) {
+        request.addHeader(name, value);
+        return this;
+    }
+
+    public Socket build() {
+        return Socket.init(httpClient, request.build());
+    }
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/SocketEvents.java

@@ -0,0 +1,76 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+import okhttp3.Response;
+
+final class SocketEvents {
+
+    final public static class MessageEvent {
+        String message;
+
+        MessageEvent(String message) {
+            this.message = message;
+        }
+    }
+
+    final public static class BaseMessageEvent {
+        String name;
+
+        BaseMessageEvent(String name) {
+            this.name = name;
+        }
+    }
+
+    final public static class ResponseMessageEvent {
+        String name;
+        String data;
+
+        ResponseMessageEvent(String name, String data) {
+            this.name = name;
+            this.data = data;
+        }
+    }
+
+    final public static class OpenStatusEvent {
+        Response response;
+
+        OpenStatusEvent(Response response) {
+            this.response = response;
+        }
+    }
+
+    final public static class CloseStatusEvent {
+        int code;
+        String reason;
+
+        CloseStatusEvent(int code, String reason) {
+            this.code = code;
+            this.reason = reason;
+        }
+    }
+
+    final public static class FailureStatusEvent {
+        Throwable throwable;
+
+        FailureStatusEvent(Throwable throwable) {
+            this.throwable = throwable;
+        }
+    }
+
+    final public static class ReconnectStatusEvent {
+        int attemptsCount;
+        long attemptDelay;
+
+        ReconnectStatusEvent(int attemptsCount, long attemptDelay) {
+            this.attemptsCount = attemptsCount;
+            this.attemptDelay = attemptDelay;
+        }
+    }
+
+    final public static class ChangeStatusEvent {
+        SocketState status;
+
+        ChangeStatusEvent(SocketState status) {
+            this.status = status;
+        }
+    }
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/SocketListenersBinder.java

@@ -0,0 +1,123 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+
+import androidx.annotation.NonNull;
+
+/**
+ * Class to add listeners to specific activity/fragment
+ */
+public class SocketListenersBinder {
+
+    private Object mLifecycle;
+    private Socket mSocket;
+
+    SocketListenersBinder(Object lifecycle, Socket socket) {
+        mLifecycle = lifecycle;
+        mSocket = socket;
+    }
+
+    /**
+     * Set state listener which fired every time {@link Socket#mState} changed.
+     *
+     * @param listener state change listener
+     */
+    public SocketListenersBinder addOnChangeStateListener(@NonNull OnStateChangeListener listener) {
+        // OpenStatusEvent
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.OpenStatusEvent.class,
+                e -> listener.onOpen(e.response), listener);
+
+        // CloseStatusEvent
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.CloseStatusEvent.class,
+                e -> listener.onClosed(e.code, e.reason), listener);
+
+        // FailureStatusEvent
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.FailureStatusEvent.class,
+                e -> listener.onFailure(e.throwable), listener);
+
+        // ReconnectStatusEvent
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.ReconnectStatusEvent.class,
+                e -> listener.onReconnect(e.attemptsCount, e.attemptDelay), listener);
+
+        // ChangeStatusEvent
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.ChangeStatusEvent.class,
+                e -> listener.onChange(e.status), listener);
+        return this;
+    }
+
+    /**
+     * Message listener will be called in any message received even if it's not
+     * in a {event -> data} format.
+     *
+     * @param listener message listener
+     */
+    public SocketListenersBinder addOnMessageListener(@NonNull OnMessageListener listener) {
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.MessageEvent.class, e -> listener.onMessage(e.message), listener);
+        return this;
+    }
+
+    /**
+     * Set listener which fired every time message received with contained data.
+     *
+     * @param listener message on arrive listener
+     */
+    public SocketListenersBinder addOnEventListener(@NonNull String event, @NonNull OnEventListener listener) {
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.BaseMessageEvent.class, e -> {
+            if (!event.equals(e.name)) return; // skip if not same event name
+            listener.onMessage(e.name);
+        }, listener);
+        return this;
+    }
+
+    /**
+     * Set listener which fired every time message received with contained data.
+     *
+     * @param listener message on arrive listener
+     */
+    public SocketListenersBinder setOnEventResponseListener(@NonNull String event, @NonNull OnEventResponseListener listener) {
+        mSocket.addEventSubscription(mLifecycle, SocketEvents.ResponseMessageEvent.class, e -> {
+            if (!event.equals(e.name)) return; // skip if not same event name
+            listener.onMessage(e.name, e.data);
+        }, listener);
+        return this;
+    }
+
+    /**
+     * Remove listener from being receive new calls.
+     *
+     * @param listener message on arrive listener
+     */
+    public void removeListener(@NonNull OnStateChangeListener listener) {
+        // remove listeners
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.OpenStatusEvent.class, listener);
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.CloseStatusEvent.class, listener);
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.FailureStatusEvent.class, listener);
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.ReconnectStatusEvent.class, listener);
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.ChangeStatusEvent.class, listener);
+    }
+
+    /**
+     * Remove listener from being receive new calls.
+     *
+     * @param listener listener to be deleted
+     */
+    public void removeListener(@NonNull OnMessageListener listener) {
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.MessageEvent.class, listener);
+    }
+
+    /**
+     * Remove listener from being receive new calls.
+     *
+     * @param listener listener to be deleted
+     */
+    public void removeListener(@NonNull OnEventListener listener) {
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.BaseMessageEvent.class, listener);
+    }
+
+    /**
+     * Remove listener from being receive new calls.
+     *
+     * @param listener listener to be deleted
+     */
+    public void removeListener(@NonNull OnEventResponseListener listener) {
+        mSocket.removeEventSubscriptionListener(mLifecycle, SocketEvents.ResponseMessageEvent.class, listener);
+    }
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTC/websocket/SocketState.java

@@ -0,0 +1,7 @@
+package sq.rogue.rosettadrone.plugins.WebRTC.websocket;
+/**
+ * Main socket connection states
+ */
+public enum SocketState {
+    CLOSED, CLOSING, CONNECT_ERROR, RECONNECT_ATTEMPT, RECONNECTING, OPENING, OPEN
+}

app/src/main/java/sq/rogue/rosettadrone/plugins/WebRTCStreaming.java

@@ -0,0 +1,112 @@
+package sq.rogue.rosettadrone.plugins;
+
+import android.util.Log;
+import java.util.concurrent.TimeUnit;
+
+import sq.rogue.rosettadrone.Plugin;
+import sq.rogue.rosettadrone.RDApplication;
+import sq.rogue.rosettadrone.plugins.WebRTC.DJIStreamer;
+import sq.rogue.rosettadrone.plugins.WebRTC.websocket.Socket;
+import sq.rogue.rosettadrone.plugins.WebRTC.websocket.SocketBuilder;
+import sq.rogue.rosettadrone.plugins.WebRTC.websocket.OnStateChangeListener;
+import sq.rogue.rosettadrone.plugins.WebRTC.websocket.SocketState;
+
+public class WebRTCStreaming extends Plugin {
+    private static final String TAG = WebRTCStreaming.class.getSimpleName();
+    private DJIStreamer djiStreamer;
+    private Socket mSocket;
+    private static final boolean TEST = false; // Send a testing stream
+    private String stunServer;
+
+    public void initWebSocket() {
+        Log.d(TAG, "initWebSocket");
+        stunServer = getPrefString("pref_webrtc_stun_server", "stun:stun.l.google.com:19302");
+        // init websocket
+        mSocket = SocketBuilder
+                .with(getPrefString("pref_webrtc_signaling_server", "ws://192.168.1.220:8090"))
+                .setPingInterval(5, TimeUnit.SECONDS)
+                .build();
+
+        // add ws states listeners
+        mSocket.addOnChangeStateListener(new OnStateChangeListener() {
+            // Socket connection events
+            @Override
+            public void onChange(SocketState status) {
+                switch (status) {
+                    case OPEN:
+                        // new OnlineEvent();
+                        break;
+                    case CLOSING: case CLOSED: case RECONNECTING:
+                    case RECONNECT_ATTEMPT: case CONNECT_ERROR:
+                        // new OfflineEvent();
+                        break;
+                }
+            }
+            @Override
+            public void onClosed(int code, String reason) {
+                // socket should be always connected
+                // Even it's closed, open the connection again
+                mSocket.connect();
+            }
+        });
+        mSocket.connect();
+    }
+
+    public void initStreaming() {
+        if(TEST || RDApplication.isTestMode) {
+            // TODO: fake video streaming?
+        }
+        else {
+            if (pluginManager.mainActivity.mModel.m_model == null) {
+                String msg = "Couldn't get model. Reconnect or restart app.";
+                Log.e(TAG, msg);
+                pluginManager.mainActivity.logMessageDJI(msg);
+                pluginManager.mainActivity.finish();
+            }
+            else if (djiStreamer == null) {
+                djiStreamer = new DJIStreamer(pluginManager.mainActivity,
+                        pluginManager.mainActivity.mModel.m_model, stunServer);
+            }
+        }
+    }
+
+    public void start() {
+        if (getPrefBoolean("pref_enable_webrtc", false)) {
+            pluginManager.mainActivity.useCustomDecoder = false; // Messes up the buffer received by onYuvDataReceived()
+            pluginManager.mainActivity.useOutputSurface = false; // Avoid crash when clicking on minimap
+            initWebSocket();
+        }
+    }
+
+    public void resume() {
+        if (getPrefBoolean("pref_enable_webrtc", false)) {
+            initStreaming();
+        }
+    }
+
+    public void pause() {
+        if (getPrefBoolean("pref_enable_webrtc", false) && djiStreamer != null) {
+            djiStreamer.closeVideoStream();
+        }
+    }
+
+    public void onVideoChange() {
+        // TODO: stop/start connections?
+        Log.d(TAG, "onVideoChange");
+    }
+
+    public void stop() {
+        if(TEST || RDApplication.isTestMode) {
+            // TODO: stop fake video stream;
+        } else {
+            pause();
+        }
+
+        if (getPrefBoolean("pref_enable_webrtc", false) && mSocket != null)
+            mSocket.terminate();
+    }
+
+    public boolean isEnabled() {
+        return true;
+    }
+}

app/src/main/res/values/strings.xml

@@ -142,6 +142,12 @@
     <!-- Logs -->
     <string name="pref_log_messages">Log MAVLink messages</string>
 
+    <!-- WebRTC -->
+    <string name="webrtc_settings">WebRTC Video</string>
+    <string name="pref_enable_webrtc">Enable WebRTC video streaming</string>
+    <string name="pref_webrtc_signaling_address">Signaling server address</string>
+    <string name="pref_webrtc_stun_address">STUN server address</string>
+
     <!-- Communication Channels -->
     <string name="dji">DJI</string>
     <string name="gcs">GCS</string>

app/src/main/res/xml/preferences.xml

@@ -319,6 +319,28 @@
 
     </PreferenceCategory>
 
+    <PreferenceCategory
+        android:key="webrtc_prefs"
+        android:title="@string/webrtc_settings">
+
+        <androidx.preference.SwitchPreferenceCompat
+            android:key="pref_enable_webrtc"
+            android:title="@string/pref_enable_webrtc"
+            android:defaultValue="false" />
+
+        <androidx.preference.EditTextPreference
+            android:defaultValue="ws://127.0.0.1:8090"
+            android:dependency="pref_enable_webrtc"
+            android:key="pref_webrtc_signaling_server"
+            android:title="@string/pref_webrtc_signaling_address" />
+        <androidx.preference.EditTextPreference
+            android:defaultValue="stun:stun.l.google.com:19302"
+            android:dependency="pref_enable_webrtc"
+            android:key="pref_webrtc_stun_server"
+            android:title="@string/pref_webrtc_stun_address" />
+
+    </PreferenceCategory>
+
     <PreferenceCategory
         android:key="log_prefs"
         android:title="@string/log_settings">