android/aitt/src/main/java/com/samsung/android/aitt/stream/RTSPStream.java

   1 /*
   2  * Copyright (c) 2022 Samsung Electronics Co., Ltd All Rights Reserved
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package com.samsung.android.aitt.stream;
  17
  18 import android.util.Log;
  19
  20 import com.google.flatbuffers.FlexBuffers;
  21 import com.google.flatbuffers.FlexBuffersBuilder;
  22 import com.samsung.android.aitt.internal.Definitions;
  23 import com.samsung.android.aittnative.JniInterface;
  24 import com.samsung.android.modules.rtsp.RTSPClient;
  25
  26 import java.nio.ByteBuffer;
  27 import java.util.HashMap;
  28 import java.util.Map;
  29 import java.util.concurrent.atomic.AtomicBoolean;
  30
  31 /**
  32  * Class to implement RTSPStream functionalities
  33  */
  34 public class RTSPStream implements AittStream {
  35
  36     private static final String TAG = "RTSPStream";
  37     private static final String SERVER_STATE = "server_state";
  38     private static final String URL = "url";
  39     private static final String ID = "id";
  40     private static final String PASSWORD = "password";
  41     private static final String URL_PREFIX = "rtsp://";
  42     private static final String HEIGHT = "height";
  43     private static final String WIDTH = "width";
  44
  45     private final StreamRole streamRole;
  46     private final String topic;
  47     private static int streamHeight;
  48     private static int streamWidth;
  49     private final DiscoveryInfo discoveryInfo = new DiscoveryInfo();
  50     private RTSPClient rtspClient;
  51     private StreamDataCallback streamCallback;
  52     private StreamStateChangeCallback stateChangeCallback = null;
  53     private JniInterface jniInterface = null;
  54     private StreamState serverState = StreamState.INIT;
  55     private StreamState clientState = StreamState.INIT;
  56
  57     /**
  58      * RTSPStream constructor
  59      *
  60      * @param topic      Topic to which streaming is invoked
  61      * @param streamRole Role of the RTSPStream object
  62      */
  63     private RTSPStream(String topic, StreamRole streamRole) {
  64         this.topic = topic;
  65         this.streamRole = streamRole;
  66
  67         if (streamRole == StreamRole.SUBSCRIBER) {
  68             RTSPClient.ReceiveDataCallback dataCallback = frame -> {
  69                 if (streamCallback != null)
  70                     streamCallback.pushStreamData(frame);
  71             };
  72
  73             rtspClient = new RTSPClient(new AtomicBoolean(false), dataCallback);
  74         }
  75     }
  76
  77     /**
  78      * Class to store discovery parameters in an object
  79      */
  80     private static class DiscoveryInfo {
  81         private String url;
  82         private String id;
  83         private String password;
  84         private int height;
  85         private int width;
  86     }
  87
  88     /**
  89      * Create and return RTSPStream object for subscriber role
  90      *
  91      * @param topic      Topic to which Subscribe role is set
  92      * @param streamRole Role of the RTSPStream object
  93      * @return RTSPStream object
  94      */
  95     public static RTSPStream createSubscriberStream(String topic, StreamRole streamRole) {
  96         if (streamRole != StreamRole.SUBSCRIBER)
  97             throw new IllegalArgumentException("The role of this stream is not subscriber.");
  98
  99         return new RTSPStream(topic, streamRole);
 100     }
 101
 102     /**
 103      * Create and return RTSPStream object for publisher role
 104      *
 105      * @param topic      Topic to which Publisher role is set
 106      * @param streamRole Role of the RTSPStream object
 107      * @return RTSPStream object
 108      */
 109     public static RTSPStream createPublisherStream(String topic, StreamRole streamRole) {
 110         if (streamRole != StreamRole.PUBLISHER)
 111             throw new IllegalArgumentException("The role of this stream is not publisher.");
 112
 113         return new RTSPStream(topic, streamRole);
 114     }
 115
 116     /**
 117      * Method to set configuration
 118      *
 119      * @param config AittStreamConfig object
 120      */
 121     @Override
 122     public void setConfig(AittStreamConfig config) {
 123         if (config == null)
 124             throw new IllegalArgumentException("Invalid configuration");
 125
 126         if (config.getUrl() != null) {
 127             String url = config.getUrl();
 128             if (!url.startsWith(URL_PREFIX))
 129                 throw new IllegalArgumentException("Invalid RTSP URL");
 130
 131             discoveryInfo.url = config.getUrl();
 132         }
 133         if (config.getId() != null) {
 134             discoveryInfo.id = config.getId();
 135         }
 136         if (config.getPassword() != null) {
 137             discoveryInfo.password = config.getPassword();
 138         }
 139
 140         discoveryInfo.height = config.getHeight();
 141
 142         discoveryInfo.width = config.getWidth();
 143     }
 144
 145     /**
 146      * Method to start stream
 147      */
 148     @Override
 149     public void start() {
 150         if (streamRole == StreamRole.SUBSCRIBER) {
 151             if (serverState == StreamState.READY) {
 152                 startRtspClient();
 153             } else {
 154                 Log.d(TAG, "RTSP server not yet ready");
 155                 updateState(streamRole, StreamState.READY);
 156             }
 157         } else {
 158             updateState(streamRole, StreamState.READY);
 159             updateDiscoveryMessage();
 160         }
 161     }
 162
 163     /**
 164      * Method to publish to a topic
 165      *
 166      * @param topic   String topic to which data is published
 167      * @param ip      Ip of the receiver
 168      * @param port    Port of the receiver
 169      * @param message Data to be published
 170      * @return returns status
 171      */
 172     @Override
 173     public boolean publish(String topic, String ip, int port, byte[] message) {
 174         // TODO: implement this function.
 175         return true;
 176     }
 177
 178     /**
 179      * Method to disconnect from the broker
 180      */
 181     @Override
 182     public void disconnect() {
 183         //ToDo : disconnect and stop can be merged
 184         if (jniInterface != null)
 185             jniInterface.removeDiscoveryCallback(topic);
 186     }
 187
 188     /**
 189      * Method to stop the stream
 190      */
 191     @Override
 192     public void stop() {
 193         if (streamRole == StreamRole.SUBSCRIBER) {
 194             if (clientState == StreamState.PLAYING)
 195                 rtspClient.stop();
 196             updateState(streamRole, StreamState.INIT);
 197         } else {
 198             updateState(streamRole, StreamState.INIT);
 199             updateDiscoveryMessage();
 200         }
 201     }
 202
 203     /**
 204      * Method to pause the stream
 205      */
 206     public void pause() {
 207         // TODO: implement this function.
 208     }
 209
 210     /**
 211      * Method to record the stream
 212      */
 213     public void record() {
 214         // TODO: implement this function.
 215     }
 216
 217     /**
 218      * Method to set state callback
 219      */
 220     @Override
 221     public void setStateCallback(StreamStateChangeCallback callback) {
 222         this.stateChangeCallback = callback;
 223     }
 224
 225     /**
 226      * Method to set subscribe callback
 227      *
 228      * @param streamDataCallback subscribe callback object
 229      */
 230     @Override
 231     public void setReceiveCallback(StreamDataCallback streamDataCallback) {
 232         if (streamRole == StreamRole.SUBSCRIBER)
 233             streamCallback = streamDataCallback;
 234         else
 235             throw new IllegalArgumentException("The role of this stream is not subscriber");
 236     }
 237
 238     /**
 239      * Method to receive stream height
 240      *
 241      * @return returns height of the stream
 242      */
 243     @Override
 244     public int getStreamHeight() {
 245         return streamHeight;
 246     }
 247
 248     /**
 249      * Method to receive stream width
 250      *
 251      * @return returns width of the stream
 252      */
 253     @Override
 254     public int getStreamWidth() {
 255         return streamWidth;
 256     }
 257
 258     /**
 259      * Method to set subscribe callback
 260      *
 261      * @param jniInterface JniInterface object
 262      */
 263     public void setJNIInterface(JniInterface jniInterface) {
 264         this.jniInterface = jniInterface;
 265
 266         jniInterface.setDiscoveryCallback(topic, (status, data) -> {
 267             Log.d(TAG, "Received discovery callback");
 268             if (streamRole == StreamRole.PUBLISHER)
 269                 return;
 270
 271             if (status.compareTo(Definitions.WILL_LEAVE_NETWORK) == 0) {
 272                 if (clientState == StreamState.PLAYING) {
 273                     rtspClient.stop();
 274                     updateState(streamRole, StreamState.READY);
 275                 } else {
 276                     updateState(streamRole, StreamState.INIT);
 277                 }
 278                 return;
 279             }
 280
 281             ByteBuffer buffer = ByteBuffer.wrap(data);
 282             FlexBuffers.Map map = FlexBuffers.getRoot(buffer).asMap();
 283             if (map.size() != 6) {
 284                 Log.e(TAG, "Invalid RTSP discovery message");
 285                 return;
 286             }
 287
 288             StreamState state = StreamState.values()[map.get(SERVER_STATE).asInt()];
 289             updateState(StreamRole.PUBLISHER, state);
 290             discoveryInfo.url = map.get(URL).asString();
 291             discoveryInfo.id = map.get(ID).asString();
 292             discoveryInfo.password = map.get(PASSWORD).asString();
 293             discoveryInfo.height = map.get(HEIGHT).asInt();
 294             discoveryInfo.width = map.get(WIDTH).asInt();
 295
 296             streamHeight = discoveryInfo.height;
 297             streamWidth = discoveryInfo.width;
 298
 299             String url = createCompleteUrl();
 300             Log.d(TAG, "RTSP URL : " + url);
 301             if (url.isEmpty()) {
 302                 return;
 303             }
 304             rtspClient.setRtspUrl(url);
 305             rtspClient.setResolution(discoveryInfo.height, discoveryInfo.width);
 306
 307             if (serverState == StreamState.READY) {
 308                 if (clientState == StreamState.READY) {
 309                     startRtspClient(discoveryInfo.id, discoveryInfo.password);
 310                 }
 311             } else if (serverState == StreamState.INIT) {
 312                 if (clientState == StreamState.PLAYING) {
 313                     rtspClient.stop();
 314                     updateState(streamRole, StreamState.READY);
 315                 }
 316             }
 317         });
 318     }
 319
 320     private String createCompleteUrl() {
 321         String completeUrl = discoveryInfo.url;
 322         if (completeUrl == null || completeUrl.isEmpty())
 323             return "";
 324
 325         String id = discoveryInfo.id;
 326         if (id == null || id.isEmpty())
 327             return completeUrl;
 328
 329         String password = discoveryInfo.password;
 330         if (password == null || password.isEmpty())
 331             return completeUrl;
 332
 333         completeUrl = new StringBuilder(completeUrl).insert(URL_PREFIX.length(), id + ":" + password + "@").toString();
 334         return completeUrl;
 335     }
 336
 337     private void startRtspClient() {
 338         startRtspClient("", "");
 339     }
 340
 341     private void startRtspClient(String id, String password) {
 342         RTSPClient.SocketConnectCallback cb = socketSuccess -> {
 343             if (socketSuccess) {
 344                 updateState(streamRole, StreamState.PLAYING);
 345                 rtspClient.initRtspClient(id, password);
 346                 rtspClient.start();
 347             } else {
 348                 Log.e(TAG, "Error creating socket");
 349             }
 350         };
 351
 352         rtspClient.createClientSocket(cb);
 353     }
 354
 355     private void updateDiscoveryMessage() {
 356         FlexBuffersBuilder fbb = new FlexBuffersBuilder(ByteBuffer.allocate(512));
 357         {
 358             int smap = fbb.startMap();
 359             fbb.putInt(SERVER_STATE, serverState.ordinal());
 360             fbb.putString(URL, discoveryInfo.url);
 361             fbb.putString(ID, discoveryInfo.id);
 362             fbb.putString(PASSWORD, discoveryInfo.password);
 363             fbb.putInt(HEIGHT, discoveryInfo.height);
 364             fbb.putInt(WIDTH, discoveryInfo.width);
 365             fbb.endMap(null, smap);
 366         }
 367         ByteBuffer buffer = fbb.finish();
 368         byte[] data = new byte[buffer.remaining()];
 369         buffer.get(data, 0, data.length);
 370
 371         if (jniInterface != null)
 372             jniInterface.updateDiscoveryMessage(topic, data);
 373     }
 374
 375     private void updateState(StreamRole role, StreamState state) {
 376         if (role == StreamRole.PUBLISHER)
 377             serverState = state;
 378         else
 379             clientState = state;
 380
 381         if (role == streamRole && stateChangeCallback != null)
 382             stateChangeCallback.pushStataChange(state);
 383     }
 384 }