Stream service sample for the ArcGIS Runtime SDK for Java

StreamServiceListener.java

package com.esri.samples.realtime;

import com.esri.arcgisruntime.geometry.Geometry;
import com.esri.arcgisruntime.mapping.view.Graphic;
import com.esri.arcgisruntime.symbology.Renderer;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

/**
 * Connects to a stream service, gets its WebSocket URL, connects via WebSocket,
 * and emits graphics.
 */
public class StreamServiceListener implements WebSocket.Listener {

  /**
   * Used to pass new stream features back to a client.
   */
  public static abstract class StreamServiceCallback {

    /**
     * Called when the stream service's renderer is read.
     *
     * @param renderer the stream service's default renderer.
     */
    protected abstract void rendererAvailable(Renderer renderer);

    /**
     * Handles a new stream feature.
     *
     * @param newFeature a graphic representing a stream feature with a track ID
     * that the associated listener has not seen before.
     */
    protected abstract void newStreamFeature(Graphic newFeature);

  }

  private final String streamServiceUrl;
  private final StreamServiceCallback streamServiceCallback;
  private final Map<String, Graphic> trackIdToGraphic = new HashMap<>();
  private final CountDownLatch stopLatch = new CountDownLatch(1);

  private String trackIdFieldName;
  private URI webSocketSubscribeUri;
  private WebSocket webSocket;
  private StringBuilder multiTextBuffer;

  /**
   * Instantiates but does not start a new stream service listener. After
   * calling the constructor, call start() to start it.
   *
   * @param streamServiceUrl the stream service URL, e.g.
   * https://realtimegis.esri.com:6443/arcgis/rest/services/Helicopter/StreamServer
   * . This should be a valid non-null stream service URL; otherwise, the
   * start() method will probably throw an exception.
   * @param streamServiceCallback a callback object. It can be null, but then
   * this listener won't be able to give you anything useful.
   * @see #start()
   */
  public StreamServiceListener(String streamServiceUrl, StreamServiceCallback streamServiceCallback) {
    this.streamServiceUrl = streamServiceUrl;
    this.streamServiceCallback = streamServiceCallback;
  }

  /**
   * Tells the listener to read the stream service's JSON descriptor and connect
   * to the stream service via WebSocket.
   *
   * @throws IOException if the stream service cannot be read, either because it
   * is down or because the URL provided to the constructor is incorrect.
   * @throws URISyntaxException if the stream service URL given to the
   * constructor is syntactically incorrect.
   */
  public void start() throws IOException, URISyntaxException {
    JSONObject streamServiceJsonDescriptor = readStreamServiceJsonDescriptor();
    trackIdFieldName = readTrackIdFieldName(streamServiceJsonDescriptor);
    if (null != streamServiceCallback) {
      Renderer renderer = readRenderer(streamServiceJsonDescriptor);
      if (null != renderer) {
        streamServiceCallback.rendererAvailable(renderer);
      }
    }
    webSocketSubscribeUri = readWebSocketSubscribeUri(streamServiceJsonDescriptor);
    new Thread(() -> {
      try {
        HttpClient client = HttpClient.newHttpClient();
        WebSocket.Builder builder = client.newWebSocketBuilder();
        webSocket = builder.buildAsync(webSocketSubscribeUri, StreamServiceListener.this).join();
        stopLatch.await();
      } catch (InterruptedException ex) {
        Logger.getLogger(StreamServiceListener.class.getName()).log(Level.SEVERE, null, ex);
      }
    }).start();
  }

  private JSONObject readStreamServiceJsonDescriptor() throws IOException {
    StringBuilder sb = new StringBuilder();
    try (BufferedReader in = new BufferedReader(new InputStreamReader(
        new URL(streamServiceUrl + "?f=json").openStream()
    ))) {
      String line;
      while (null != (line = in.readLine())) {
        sb.append(line);
      }
    }
    return new JSONObject(sb.toString());
  }

  private static String readTrackIdFieldName(JSONObject streamServiceJsonDescriptor) {
    String trackIdFieldName = null;
    if (streamServiceJsonDescriptor.has("timeInfo")) {
      JSONObject timeInfoObj = streamServiceJsonDescriptor.getJSONObject("timeInfo");
      if (timeInfoObj.has("trackIdField")) {
        trackIdFieldName = timeInfoObj.getString("trackIdField");
      }
    }
    return trackIdFieldName;
  }

  private static Renderer readRenderer(JSONObject streamServiceJsonDescriptor) {
    if (streamServiceJsonDescriptor.has("drawingInfo")) {
      JSONObject drawingInfo = streamServiceJsonDescriptor.getJSONObject("drawingInfo");
      if (drawingInfo.has("renderer")) {
        return Renderer.fromJson(drawingInfo.getJSONObject("renderer").toString());
      }
    }
    return null;
  }

  private static URI readWebSocketSubscribeUri(JSONObject streamServiceJsonDescriptor) throws IOException, URISyntaxException {
    if (streamServiceJsonDescriptor.has("streamUrls")) {
      JSONArray streamUrls = streamServiceJsonDescriptor.getJSONArray("streamUrls");
      if (0 < streamUrls.length()) {
        JSONObject streamUrlObj = streamUrls.getJSONObject(0);
        if (streamUrlObj.has("urls")) {
          JSONArray urls = streamUrlObj.getJSONArray("urls");
          if (0 < urls.length()) {
            return new URI(urls.getString(0) + "/subscribe");
          } else {
            throw new IOException("Stream service JSON descriptor's urls array is empty");
          }
        } else {
          throw new IOException("Stream service JSON descriptor does not contain urls");
        }
      } else {
        throw new IOException("Stream service JSON descriptor's streamUrls array is empty");
      }
    } else {
      throw new IOException("Stream service JSON descriptor does not contain streamUrls");
    }
  }

  @Override
  public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
    String json = data.toString();
    if (last) {
      if (null != streamServiceCallback) {
        if (null != multiTextBuffer) {
          multiTextBuffer.append(json);
          json = multiTextBuffer.toString();
        }
        try {
          JSONObject obj = new JSONObject(json);
          createOrUpdateGraphic(
              Geometry.fromJson(obj.getJSONObject("geometry").toString()),
              obj.getJSONObject("attributes").toMap()
          );
        } catch (JSONException ex) {
          Logger.getLogger(App.class.getName())
              .log(Level.SEVERE, String.format("JSONException related to this string (last is %b): %s", last, json), ex);
        }
      }
      multiTextBuffer = null;
    } else {
      if (null == multiTextBuffer) {
        multiTextBuffer = new StringBuilder(json);
      } else {
        multiTextBuffer.append(json);
      }
    }
    return WebSocket.Listener.super.onText(webSocket, data, last);
  }

  private void createOrUpdateGraphic(Geometry geometry, Map<String, Object> attributes) {
    Graphic graphic = null;
    String trackId = null;
    if (null != trackIdFieldName) {
      Object trackIdObj = attributes.get(trackIdFieldName);
      if (null != trackIdObj) {
        trackId = trackIdObj.toString();
        graphic = trackIdToGraphic.get(trackId);
      }
    }
    boolean newGraphic = false;
    if (null == graphic) {
      graphic = new Graphic();
      newGraphic = true;
      if (null != trackId) {
        trackIdToGraphic.put(trackId, graphic);
      }
    }
    graphic.getAttributes().putAll(attributes);
    graphic.setGeometry(geometry);
    if (newGraphic) {
      streamServiceCallback.newStreamFeature(graphic);
    }
  }

  @Override
  public void onError(WebSocket webSocket, Throwable error) {
    // TODO do something to notify the client code and/or the user
    System.out.println("onError: " + error.getMessage());
    error.printStackTrace();
    WebSocket.Listener.super.onError(webSocket, error);
  }

  /**
   * Closes the WebSocket connection. You can start it again by calling start().
   */
  public void close() {
    webSocket.sendClose(200, "OK");
    stopLatch.countDown();
  }

  @Override
  public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
    stopLatch.countDown();
    return WebSocket.Listener.super.onClose(webSocket, statusCode, reason);
  }

}

StreamServiceSample.java

package com.esri.samples.realtime;

import javafx.application.Application;
import javafx.scene.Scene;
import javafx.scene.layout.StackPane;
import javafx.stage.Stage;

import com.esri.arcgisruntime.mapping.ArcGISMap;
import com.esri.arcgisruntime.mapping.Basemap;
import com.esri.arcgisruntime.mapping.view.Graphic;
import com.esri.arcgisruntime.mapping.view.GraphicsOverlay;
import com.esri.arcgisruntime.mapping.view.MapView;
import com.esri.arcgisruntime.symbology.Renderer;

public class StreamServiceSample extends Application {

  private static final String STREAM_SERVICE_URL
      = "http://ec2-75-101-155-202.compute-1.amazonaws.com:6080/arcgis/rest/services/ASDITrackInformation/StreamServer";
//      = "https://realtimegis.esri.com:6443/arcgis/rest/services/Helicopter/StreamServer";
//      = "https://publicsafetygesdev.bd.esri.com/server/rest/services/Race_Leader_AVL/StreamServer";
//      = "https://gis.yakimawa.gov/geoevent/rest/services/avl-service-out/StreamServer";
//      = "https://gis.yakimawa.gov/geoevent/rest/services/ali-service-out/StreamServer";
//      = "https://gis.yakimawa.gov/geoevent/rest/services/refuse-service-out/StreamServer";
//      = "https://logisgeoevent.loudoun.gov/ge/rest/services/WazeAlert-Flat/StreamServer";
//      = "https://logisgeoevent.loudoun.gov/ge/rest/services/WazeJam-Flat/StreamServer";
//      = "https://geoeventsample1.esri.com:6443/arcgis/rest/services/LABus/StreamServer";
//      = "https://gisags60.lincoln.ne.gov/arcgis/rest/services/LTU/StreamSnowPlows/StreamServer";
//      = "https://dotscoazgis1.dot.state.fl.us/arcgis/rest/services/ISS_Location/StreamServer";

  private MapView mapView;

  private final GraphicsOverlay streamGraphicsOverlay = new GraphicsOverlay();
  private final StreamServiceListener streamServiceListener;

  public StreamServiceSample() {
    StreamServiceListener.StreamServiceCallback callback = new StreamServiceListener.StreamServiceCallback() {
      @Override
      protected void rendererAvailable(Renderer renderer) {
        streamGraphicsOverlay.setRenderer(renderer);
      }

      @Override
      protected void newStreamFeature(Graphic newFeature) {
        streamGraphicsOverlay.getGraphics().add(newFeature);
      }
    };
    this.streamServiceListener = new StreamServiceListener(STREAM_SERVICE_URL, callback);
  }

  @Override
  public void start(Stage stage) {

    try {
      // create stack pane and application scene
      StackPane stackPane = new StackPane();
      Scene scene = new Scene(stackPane);

      // set title, size, and add scene to stage
      stage.setTitle("Display Map Sample");
      stage.setWidth(800);
      stage.setHeight(700);
      stage.setScene(scene);
      stage.show();

      // create a ArcGISMap with the a Basemap instance with an Imagery base
      // layer
      ArcGISMap map = new ArcGISMap(Basemap.createImagery());

      // set the map to be displayed in this view
      mapView = new MapView();
      mapView.setMap(map);

      mapView.getGraphicsOverlays().add(streamGraphicsOverlay);

      // add the map view to stack pane
      stackPane.getChildren().addAll(mapView);

      streamServiceListener.start();
    } catch (Exception e) {
      // on any error, display the stack trace.
      e.printStackTrace();
    }
  }

  /**
   * Stops and releases all resources used in application.
   */
  @Override
  public void stop() {
    streamServiceListener.close();

    if (mapView != null) {
      mapView.dispose();
    }
  }

  /**
   * Opens and runs application.
   *
   * @param args arguments passed to this application
   */
  public static void main(String[] args) {

    Application.launch(args);
  }

}