NtripClient.java

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

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.Authenticator;
import java.net.HttpURLConnection;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.net.Proxy.Type;
import java.net.SocketAddress;
import java.net.URL;
import java.net.URLConnection;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Formatter;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.orekit.errors.OrekitException;
import org.orekit.errors.OrekitMessages;
import org.orekit.gnss.metric.messages.ParsedMessage;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    /** Build a client for NTRIP.
     * <p>
     * The default configuration uses default timeout, default reconnection
     * parameters, no GPS fix and no proxy.
     * </p>
     * @param host caster host providing the source table
     * @param port port to use for connection
     * see {@link #DEFAULT_PORT}
     */
    public NtripClient(final String host, final int port) {
        this.host         = host;
        this.port         = port;
        this.observers    = new ArrayList<>();
        this.monitors     = new HashMap<>();
        setTimeout(DEFAULT_TIMEOUT);
        setReconnectParameters(DEFAULT_RECONNECT_DELAY,
                               DEFAULT_RECONNECT_DELAY_FACTOR,
                               DEFAULT_MAX_RECONNECT);
        setProxy(Type.DIRECT, null, -1);
        this.gga             = new AtomicReference<String>(null);
        this.sourceTable     = null;
        this.executorService = null;
    }

    /** Get the caster host.
     * @return caster host
     */
    public String getHost() {
        return host;
    }

    /** Get the port to use for connection.
     * @return port to use for connection
     */
    public int getPort() {
        return port;
    }

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

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

    /** Set proxy parameters.
     * @param type proxy type
     * @param proxyHost host name of the proxy (ignored if {@code type} is {@code Proxy.Type.DIRECT})
     * @param proxyPort port number of the proxy (ignored if {@code type} is {@code Proxy.Type.DIRECT})
     */
    public void setProxy(final Proxy.Type type, final String proxyHost, final int proxyPort) {
        try {
            if (type == Proxy.Type.DIRECT) {
                // disable proxy
                proxy = Proxy.NO_PROXY;
            } else {
                // enable proxy
                final InetAddress   hostAddress  = InetAddress.getByName(proxyHost);
                final SocketAddress proxyAddress = new InetSocketAddress(hostAddress, proxyPort);
                proxy = new Proxy(type, proxyAddress);
            }
        } catch (UnknownHostException uhe) {
            throw new OrekitException(uhe, OrekitMessages.UNKNOWN_HOST, proxyHost);
        }
    }

    /** Get proxy.
     * @return proxy to use
     */
    public Proxy getProxy() {
        return proxy;
    }

    /** Set GPS fix data to send as NMEA sentence to Ntrip caster if required.
     * @param hour hour of the fix (UTC time)
     * @param minute minute of the fix (UTC time)
     * @param second second of the fix (UTC time)
     * @param latitude latitude (radians)
     * @param longitude longitude (radians)
     * @param ellAltitude altitude above ellipsoid (m)
     * @param undulation height of the geoid above ellipsoid (m)
     */
    public void setFix(final int hour, final int minute, final double second,
                       final double latitude, final double longitude, final double ellAltitude,
                       final double undulation) {

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

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

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

            // dummy values
            final int    fixQuality = 1;
            final int    nbSat      = 4;
            final double hdop       = 1.0;

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

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

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

    }

    /** Get NMEA GGA sentence.
     * @return NMEA GGA sentence (may be null)
     */
    String getGGA() {
        return gga.get();
    }

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

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

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

    }

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

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

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

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

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

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

                        ++lineNumber;
                        line = line.trim();
                        if (line.length() == 0) {
                            continue;
                        }

                        if (line.startsWith(RecordType.CAS.toString())) {
                            table.addCasterRecord(new CasterRecord(line));
                        } else if (line.startsWith(RecordType.NET.toString())) {
                            table.addNetworkRecord(new NetworkRecord(line));
                        } else if (line.startsWith(RecordType.STR.toString())) {
                            table.addDataStreamRecord(new DataStreamRecord(line));
                        } else if (line.startsWith("ENDSOURCETABLE")) {
                            // we have reached end of table
                            break;
                        } else {
                            throw new OrekitException(OrekitMessages.SOURCETABLE_PARSE_ERROR,
                                                      connection.getURL().getHost(), lineNumber, line);
                        }

                    }
                }

                sourceTable = table;
                return table;

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

        return sourceTable;

    }

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

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

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

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

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

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

    }

    /** Check if any of the streaming thread has thrown an exception.
     * <p>
     * If a streaming thread has thrown an exception, it will be rethrown here
     * </p>
     */
    public void checkException() {
        // check if any of the stream got an exception
        for (final  Map.Entry<String, StreamMonitor> entry : monitors.entrySet()) {
            final OrekitException exception = entry.getValue().getException();
            if (exception != null) {
                throw exception;
            }
        }
    }

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

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

        try {
            // wait for proper ending
            executorService.awaitTermination(time, TimeUnit.MILLISECONDS);
        } catch (InterruptedException ie) {
            // Restore interrupted state...
            Thread.currentThread().interrupt();
        }

        checkException();

    }

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

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

        // common headers
        connection.setRequestProperty(HOST_HEADER_KEY,       host);
        connection.setRequestProperty(VERSION_HEADER_KEY,    VERSION_HEADER_VALUE);
        connection.setRequestProperty(USER_AGENT_HEADER_KEY, USER_AGENT_HEADER_VALUE);
        connection.setRequestProperty(CONNECTION_HEADER_KEY, CONNECTION_HEADER_VALUE);

        return connection;

    }

    /** Get an header from a response.
     * @param connection connection to analyze
     * @param key header key
     * @return header value
     */
    private String getHeaderValue(final URLConnection connection, final String key) {
        final String value = connection.getHeaderField(key);
        if (value == null) {
            throw new OrekitException(OrekitMessages.MISSING_HEADER,
                                      connection.getURL().getHost(), key);
        }
        return value;
    }

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

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

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

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

        /** Simple constructor.
         * @param typeCode code for the message type
         * @param mountPoint mountPoint from which data must come (if null, notification
         * will be triggered regardless of mount point)
         * @param observer observer for this message type
         */
        ObserverHolder(final int typeCode, final String mountPoint,
                            final MessageObserver observer) {
            this.typeCode   = typeCode;
            this.mountPoint = mountPoint;
            this.observer   = observer;
        }

    }

}