NtripClient.java

  1. /* Copyright 2002-2022 CS GROUP
  2.  * Licensed to CS GROUP (CS) under one or more
  3.  * contributor license agreements.  See the NOTICE file distributed with
  4.  * this work for additional information regarding copyright ownership.
  5.  * CS licenses this file to You under the Apache License, Version 2.0
  6.  * (the "License"); you may not use this file except in compliance with
  7.  * the License.  You may obtain a copy of the License at
  8.  *
  9.  *   http://www.apache.org/licenses/LICENSE-2.0
  10.  *
  11.  * Unless required by applicable law or agreed to in writing, software
  12.  * distributed under the License is distributed on an "AS IS" BASIS,
  13.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14.  * See the License for the specific language governing permissions and
  15.  * limitations under the License.
  16.  */
  17. package org.orekit.gnss.metric.ntrip;

  18. import java.io.BufferedReader;
  19. import java.io.IOException;
  20. import java.io.InputStream;
  21. import java.io.InputStreamReader;
  22. import java.net.Authenticator;
  23. import java.net.HttpURLConnection;
  24. import java.net.InetAddress;
  25. import java.net.InetSocketAddress;
  26. import java.net.Proxy;
  27. import java.net.Proxy.Type;
  28. import java.net.SocketAddress;
  29. import java.net.URL;
  30. import java.net.URLConnection;
  31. import java.net.UnknownHostException;
  32. import java.nio.charset.StandardCharsets;
  33. import java.util.ArrayList;
  34. import java.util.Formatter;
  35. import java.util.HashMap;
  36. import java.util.List;
  37. import java.util.Locale;
  38. import java.util.Map;
  39. import java.util.concurrent.ExecutorService;
  40. import java.util.concurrent.Executors;
  41. import java.util.concurrent.TimeUnit;
  42. import java.util.concurrent.atomic.AtomicReference;

  43. import org.hipparchus.util.FastMath;
  44. import org.orekit.errors.OrekitException;
  45. import org.orekit.errors.OrekitMessages;
  46. import org.orekit.gnss.metric.messages.ParsedMessage;

  47. /** Source table for ntrip streams retrieval.
  48.  * <p>
  49.  * Note that all authentication is performed automatically by just
  50.  * calling the standard {@link Authenticator#setDefault(Authenticator)}
  51.  * method to set up an authenticator.
  52.  * </p>
  53.  * @author Luc Maisonobe
  54.  * @since 11.0
  55.  */
  56. public class NtripClient {

  57.     /** Default timeout for connections and reads (ms). */
  58.     public static final int DEFAULT_TIMEOUT = 10000;

  59.     /** Default port for ntrip communication. */
  60.     public static final int DEFAULT_PORT = 2101;

  61.     /** Default delay before we reconnect after connection close (s). */
  62.     public static final double DEFAULT_RECONNECT_DELAY = 1.0;

  63.     /** Default factor by which reconnection delay is multiplied after each attempt. */
  64.     public static final double DEFAULT_RECONNECT_DELAY_FACTOR = 1.5;

  65.     /** Default maximum number of reconnect a attempts without readin any data. */
  66.     public static final int DEFAULT_MAX_RECONNECT = 20;

  67.     /** Host header. */
  68.     private static final String HOST_HEADER_KEY = "Host";

  69.     /** User-agent header key. */
  70.     private static final String USER_AGENT_HEADER_KEY = "User-Agent";

  71.     /** User-agent header value. */
  72.     private static final String USER_AGENT_HEADER_VALUE = "NTRIP orekit/11.0";

  73.     /** Version header key. */
  74.     private static final String VERSION_HEADER_KEY = "Ntrip-Version";

  75.     /** Version header value. */
  76.     private static final String VERSION_HEADER_VALUE = "Ntrip/2.0";

  77.     /** Connection header key. */
  78.     private static final String CONNECTION_HEADER_KEY = "Connection";

  79.     /** Connection header value. */
  80.     private static final String CONNECTION_HEADER_VALUE = "close";

  81.     /** Flags header key. */
  82.     private static final String FLAGS_HEADER_KEY = "Ntrip-Flags";

  83.     /** Content type for source table. */
  84.     private static final String SOURCETABLE_CONTENT_TYPE = "gnss/sourcetable";

  85.     /** Degrees to arc minutes conversion factor. */
  86.     private static final double DEG_TO_MINUTES = 60.0;

  87.     /** Caster host. */
  88.     private final String host;

  89.     /** Caster port. */
  90.     private final int port;

  91.     /** Delay before we reconnect after connection close. */
  92.     private double reconnectDelay;

  93.     /** Multiplication factor for reconnection delay. */
  94.     private double reconnectDelayFactor;

  95.     /** Max number of reconnections. */
  96.     private int maxRetries;

  97.     /** Timeout for connections and reads. */
  98.     private int timeout;

  99.     /** Proxy to use. */
  100.     private Proxy proxy;

  101.     /** NMEA GGA sentence (may be null). */
  102.     private AtomicReference<String> gga;

  103.     /** Observers for encoded messages. */
  104.     private final List<ObserverHolder> observers;

  105.     /** Monitors for data streams. */
  106.     private final Map<String, StreamMonitor> monitors;

  107.     /** Source table. */
  108.     private SourceTable sourceTable;

  109.     /** Executor for stream monitoring tasks. */
  110.     private ExecutorService executorService;

  111.     /** Build a client for NTRIP.
  112.      * <p>
  113.      * The default configuration uses default timeout, default reconnection
  114.      * parameters, no GPS fix and no proxy.
  115.      * </p>
  116.      * @param host caster host providing the source table
  117.      * @param port port to use for connection
  118.      * see {@link #DEFAULT_PORT}
  119.      */
  120.     public NtripClient(final String host, final int port) {
  121.         this.host         = host;
  122.         this.port         = port;
  123.         this.observers    = new ArrayList<>();
  124.         this.monitors     = new HashMap<>();
  125.         setTimeout(DEFAULT_TIMEOUT);
  126.         setReconnectParameters(DEFAULT_RECONNECT_DELAY,
  127.                                DEFAULT_RECONNECT_DELAY_FACTOR,
  128.                                DEFAULT_MAX_RECONNECT);
  129.         setProxy(Type.DIRECT, null, -1);
  130.         this.gga             = new AtomicReference<String>(null);
  131.         this.sourceTable     = null;
  132.         this.executorService = null;
  133.     }

  134.     /** Get the caster host.
  135.      * @return caster host
  136.      */
  137.     public String getHost() {
  138.         return host;
  139.     }

  140.     /** Get the port to use for connection.
  141.      * @return port to use for connection
  142.      */
  143.     public int getPort() {
  144.         return port;
  145.     }

  146.     /** Set timeout for connections and reads.
  147.      * @param timeout timeout for connections and reads (ms)
  148.      */
  149.     public void setTimeout(final int timeout) {
  150.         this.timeout = timeout;
  151.     }

  152.     /** Set Reconnect parameters.
  153.      * @param delay delay before we reconnect after connection close
  154.      * @param delayFactor factor by which reconnection delay is multiplied after each attempt
  155.      * @param max max number of reconnect a attempts without reading any data
  156.      */
  157.     public void setReconnectParameters(final double delay,
  158.                                        final double delayFactor,
  159.                                        final int max) {
  160.         this.reconnectDelay       = delay;
  161.         this.reconnectDelayFactor = delayFactor;
  162.         this.maxRetries           = max;
  163.     }

  164.     /** Set proxy parameters.
  165.      * @param type proxy type
  166.      * @param proxyHost host name of the proxy (ignored if {@code type} is {@code Proxy.Type.DIRECT})
  167.      * @param proxyPort port number of the proxy (ignored if {@code type} is {@code Proxy.Type.DIRECT})
  168.      */
  169.     public void setProxy(final Proxy.Type type, final String proxyHost, final int proxyPort) {
  170.         try {
  171.             if (type == Proxy.Type.DIRECT) {
  172.                 // disable proxy
  173.                 proxy = Proxy.NO_PROXY;
  174.             } else {
  175.                 // enable proxy
  176.                 final InetAddress   hostAddress  = InetAddress.getByName(proxyHost);
  177.                 final SocketAddress proxyAddress = new InetSocketAddress(hostAddress, proxyPort);
  178.                 proxy = new Proxy(type, proxyAddress);
  179.             }
  180.         } catch (UnknownHostException uhe) {
  181.             throw new OrekitException(uhe, OrekitMessages.UNKNOWN_HOST, proxyHost);
  182.         }
  183.     }

  184.     /** Get proxy.
  185.      * @return proxy to use
  186.      */
  187.     public Proxy getProxy() {
  188.         return proxy;
  189.     }

  190.     /** Set GPS fix data to send as NMEA sentence to Ntrip caster if required.
  191.      * @param hour hour of the fix (UTC time)
  192.      * @param minute minute of the fix (UTC time)
  193.      * @param second second of the fix (UTC time)
  194.      * @param latitude latitude (radians)
  195.      * @param longitude longitude (radians)
  196.      * @param ellAltitude altitude above ellipsoid (m)
  197.      * @param undulation height of the geoid above ellipsoid (m)
  198.      */
  199.     public void setFix(final int hour, final int minute, final double second,
  200.                        final double latitude, final double longitude, final double ellAltitude,
  201.                        final double undulation) {

  202.         // convert latitude
  203.         final double latDeg = FastMath.abs(FastMath.toDegrees(latitude));
  204.         final int    dLat   = (int) FastMath.floor(latDeg);
  205.         final double mLat   = DEG_TO_MINUTES * (latDeg - dLat);
  206.         final char   cLat   = latitude >= 0.0 ? 'N' : 'S';

  207.         // convert longitude
  208.         final double lonDeg = FastMath.abs(FastMath.toDegrees(longitude));
  209.         final int    dLon   = (int) FastMath.floor(lonDeg);
  210.         final double mLon   = DEG_TO_MINUTES * (lonDeg - dLon);
  211.         final char   cLon   = longitude >= 0.0 ? 'E' : 'W';

  212.         // build NMEA GGA sentence
  213.         final StringBuilder builder = new StringBuilder(82);
  214.         try (Formatter formatter = new Formatter(builder, Locale.US)) {

  215.             // dummy values
  216.             final int    fixQuality = 1;
  217.             final int    nbSat      = 4;
  218.             final double hdop       = 1.0;

  219.             // sentence body
  220.             formatter.format("$GPGGA,%02d%02d%06.3f,%02d%07.4f,%c,%02d%07.4f,%c,%1d,%02d,%3.1f,%.1f,M,%.1f,M,,",
  221.                              hour, minute, second,
  222.                              dLat, mLat, cLat, dLon, mLon, cLon,
  223.                              fixQuality, nbSat, hdop,
  224.                              ellAltitude, undulation);

  225.             // checksum
  226.             byte sum = 0;
  227.             for (int i = 1; i < builder.length(); ++i) {
  228.                 sum ^= builder.charAt(i);
  229.             }
  230.             formatter.format("*%02X", sum);

  231.         }
  232.         gga.set(builder.toString());

  233.     }

  234.     /** Get NMEA GGA sentence.
  235.      * @return NMEA GGA sentence (may be null)
  236.      */
  237.     String getGGA() {
  238.         return gga.get();
  239.     }

  240.     /** Add an observer for an encoded messages.
  241.      * <p>
  242.      * If messages of the specified type have already been retrieved from
  243.      * a stream, the observer will be immediately notified with the last
  244.      * message from each mount point (in unspecified order) as a side effect
  245.      * of being added.
  246.      * </p>
  247.      * @param typeCode code for the message type (if set to 0, notification
  248.      * will be triggered regardless of message type)
  249.      * @param mountPoint mountPoint from which data must come (if null, notification
  250.      * will be triggered regardless of mount point)
  251.      * @param observer observer for this message type
  252.      */
  253.     public void addObserver(final int typeCode, final String mountPoint,
  254.                             final MessageObserver observer) {

  255.         // store the observer for future monitored mount points
  256.         observers.add(new ObserverHolder(typeCode, mountPoint, observer));

  257.         // check if we should also add it to already monitored mount points
  258.         for (Map.Entry<String, StreamMonitor> entry : monitors.entrySet()) {
  259.             if (mountPoint == null || mountPoint.equals(entry.getKey())) {
  260.                 entry.getValue().addObserver(typeCode, observer);
  261.             }
  262.         }

  263.     }

  264.     /** Get a sourcetable.
  265.      * @return source table from the caster
  266.      */
  267.     public SourceTable getSourceTable() {
  268.         if (sourceTable == null) {
  269.             try {

  270.                 // perform request
  271.                 final HttpURLConnection connection = connect("");

  272.                 final int responseCode = connection.getResponseCode();
  273.                 if (responseCode == HttpURLConnection.HTTP_UNAUTHORIZED) {
  274.                     throw new OrekitException(OrekitMessages.FAILED_AUTHENTICATION, "caster");
  275.                 } else if (responseCode != HttpURLConnection.HTTP_OK) {
  276.                     throw new OrekitException(OrekitMessages.CONNECTION_ERROR, host, connection.getResponseMessage());
  277.                 }

  278.                 // for this request, we MUST get a source table
  279.                 if (!SOURCETABLE_CONTENT_TYPE.equals(connection.getContentType())) {
  280.                     throw new OrekitException(OrekitMessages.UNEXPECTED_CONTENT_TYPE, connection.getContentType());
  281.                 }

  282.                 final SourceTable table = new SourceTable(getHeaderValue(connection, FLAGS_HEADER_KEY));

  283.                 // parse source table records
  284.                 try (InputStream is = connection.getInputStream();
  285.                      InputStreamReader isr = new InputStreamReader(is, StandardCharsets.UTF_8);
  286.                      BufferedReader br = new BufferedReader(isr)) {
  287.                     int lineNumber = 0;
  288.                     for (String line = br.readLine(); line != null; line = br.readLine()) {

  289.                         ++lineNumber;
  290.                         line = line.trim();
  291.                         if (line.length() == 0) {
  292.                             continue;
  293.                         }

  294.                         if (line.startsWith(RecordType.CAS.toString())) {
  295.                             table.addCasterRecord(new CasterRecord(line));
  296.                         } else if (line.startsWith(RecordType.NET.toString())) {
  297.                             table.addNetworkRecord(new NetworkRecord(line));
  298.                         } else if (line.startsWith(RecordType.STR.toString())) {
  299.                             table.addDataStreamRecord(new DataStreamRecord(line));
  300.                         } else if (line.startsWith("ENDSOURCETABLE")) {
  301.                             // we have reached end of table
  302.                             break;
  303.                         } else {
  304.                             throw new OrekitException(OrekitMessages.SOURCETABLE_PARSE_ERROR,
  305.                                                       connection.getURL().getHost(), lineNumber, line);
  306.                         }

  307.                     }
  308.                 }

  309.                 sourceTable = table;
  310.                 return table;

  311.             } catch (IOException ioe) {
  312.                 throw new OrekitException(ioe, OrekitMessages.CANNOT_PARSE_SOURCETABLE, host);
  313.             }
  314.         }

  315.         return sourceTable;

  316.     }

  317.     /** Connect to a mount point and start streaming data from it.
  318.      * <p>
  319.      * This method sets up an internal dedicated thread for continuously
  320.      * monitoring data incoming from a mount point. When new complete
  321.      * {@link ParsedMessage parsed messages} becomes available, the
  322.      * {@link MessageObserver observers} that have been registered
  323.      * using {@link #addObserver(int, String, MessageObserver) addObserver()}
  324.      * method will be notified about the message.
  325.      * </p>
  326.      * <p>
  327.      * This method must be called once for each stream to monitor.
  328.      * </p>
  329.      * @param mountPoint mount point providing the stream
  330.      * @param type messages type of the mount point
  331.      * @param requiresNMEA if true, the mount point requires a NMEA GGA sentence in the request
  332.      * @param ignoreUnknownMessageTypes if true, unknown messages types are silently ignored
  333.      */
  334.     public void startStreaming(final String mountPoint, final org.orekit.gnss.metric.ntrip.Type type,
  335.                                final boolean requiresNMEA, final boolean ignoreUnknownMessageTypes) {

  336.         if (executorService == null) {
  337.             // lazy creation of executor service, with one thread for each possible data stream
  338.             executorService = Executors.newFixedThreadPool(getSourceTable().getDataStreams().size());
  339.         }

  340.         // safety check
  341.         if (monitors.containsKey(mountPoint)) {
  342.             throw new OrekitException(OrekitMessages.MOUNPOINT_ALREADY_CONNECTED, mountPoint);
  343.         }

  344.         // create the monitor
  345.         final StreamMonitor monitor = new StreamMonitor(this, mountPoint, type, requiresNMEA, ignoreUnknownMessageTypes,
  346.                                                         reconnectDelay, reconnectDelayFactor, maxRetries);
  347.         monitors.put(mountPoint, monitor);

  348.         // set up the already known observers
  349.         for (final ObserverHolder observerHolder : observers) {
  350.             if (observerHolder.mountPoint == null ||
  351.                 observerHolder.mountPoint.equals(mountPoint)) {
  352.                 monitor.addObserver(observerHolder.typeCode, observerHolder.observer);
  353.             }
  354.         }

  355.         // start streaming data
  356.         executorService.execute(monitor);

  357.     }

  358.     /** Check if any of the streaming thread has thrown an exception.
  359.      * <p>
  360.      * If a streaming thread has thrown an exception, it will be rethrown here
  361.      * </p>
  362.      */
  363.     public void checkException() {
  364.         // check if any of the stream got an exception
  365.         for (final  Map.Entry<String, StreamMonitor> entry : monitors.entrySet()) {
  366.             final OrekitException exception = entry.getValue().getException();
  367.             if (exception != null) {
  368.                 throw exception;
  369.             }
  370.         }
  371.     }

  372.     /** Stop streaming data from all connected mount points.
  373.      * <p>
  374.      * If an exception was encountered during data streaming, it will be rethrown here
  375.      * </p>
  376.      * @param time timeout for waiting underlying threads termination (ms)
  377.      */
  378.     public void stopStreaming(final int time) {

  379.         // ask all monitors to stop retrieving data
  380.         for (final  Map.Entry<String, StreamMonitor> entry : monitors.entrySet()) {
  381.             entry.getValue().stopMonitoring();
  382.         }

  383.         try {
  384.             // wait for proper ending
  385.             executorService.awaitTermination(time, TimeUnit.MILLISECONDS);
  386.         } catch (InterruptedException ie) {
  387.             // Restore interrupted state...
  388.             Thread.currentThread().interrupt();
  389.         }

  390.         checkException();

  391.     }

  392.     /** Connect to caster.
  393.      * @param mountPoint mount point (empty for getting sourcetable)
  394.      * @return performed connection
  395.      * @throws IOException if an I/O exception occurs during connection
  396.      */
  397.     HttpURLConnection connect(final String mountPoint)
  398.         throws IOException {

  399.         // set up connection
  400.         final String protocol = "http";
  401.         final URL casterURL = new URL(protocol, host, port, "/" + mountPoint);
  402.         final HttpURLConnection connection = (HttpURLConnection) casterURL.openConnection(proxy);
  403.         connection.setConnectTimeout(timeout);
  404.         connection.setReadTimeout(timeout);

  405.         // common headers
  406.         connection.setRequestProperty(HOST_HEADER_KEY,       host);
  407.         connection.setRequestProperty(VERSION_HEADER_KEY,    VERSION_HEADER_VALUE);
  408.         connection.setRequestProperty(USER_AGENT_HEADER_KEY, USER_AGENT_HEADER_VALUE);
  409.         connection.setRequestProperty(CONNECTION_HEADER_KEY, CONNECTION_HEADER_VALUE);

  410.         return connection;

  411.     }

  412.     /** Get an header from a response.
  413.      * @param connection connection to analyze
  414.      * @param key header key
  415.      * @return header value
  416.      */
  417.     private String getHeaderValue(final URLConnection connection, final String key) {
  418.         final String value = connection.getHeaderField(key);
  419.         if (value == null) {
  420.             throw new OrekitException(OrekitMessages.MISSING_HEADER,
  421.                                       connection.getURL().getHost(), key);
  422.         }
  423.         return value;
  424.     }

  425.     /** Local holder for observers. */
  426.     private static class ObserverHolder {

  427.         /** Code for the message type. */
  428.         private final int typeCode;

  429.         /** Mount point. */
  430.         private final String mountPoint;

  431.         /** Observer to notify. */
  432.         private final MessageObserver observer;

  433.         /** Simple constructor.
  434.          * @param typeCode code for the message type
  435.          * @param mountPoint mountPoint from which data must come (if null, notification
  436.          * will be triggered regardless of mount point)
  437.          * @param observer observer for this message type
  438.          */
  439.         ObserverHolder(final int typeCode, final String mountPoint,
  440.                             final MessageObserver observer) {
  441.             this.typeCode   = typeCode;
  442.             this.mountPoint = mountPoint;
  443.             this.observer   = observer;
  444.         }

  445.     }

  446. }