StreamMonitor.java

  1. /* Copyright 2002-2024 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;

  19. import java.io.IOException;
  20. import java.io.InputStream;
  21. import java.net.HttpURLConnection;
  22. import java.net.SocketTimeoutException;
  23. import java.net.URISyntaxException;
  24. import java.util.ArrayList;
  25. import java.util.Arrays;
  26. import java.util.HashMap;
  27. import java.util.List;
  28. import java.util.Map;
  29. import java.util.concurrent.atomic.AtomicBoolean;
  30. import java.util.concurrent.atomic.AtomicReference;

  32. import org.hipparchus.util.FastMath;
  33. import org.orekit.errors.OrekitException;
  34. import org.orekit.errors.OrekitInternalError;
  35. import org.orekit.errors.OrekitMessages;
  36. import org.orekit.gnss.metric.messages.ParsedMessage;
  37. import org.orekit.gnss.metric.parser.AbstractEncodedMessage;
  38. import org.orekit.gnss.metric.parser.MessagesParser;

  40. /** Monitor for retrieving streamed data from one mount point.
  41.  * @author Luc Maisonobe
  42.  * @since 11.0
  43.  */
  44. public class StreamMonitor extends AbstractEncodedMessage implements Runnable {

  46.     /** GGA header key. */
  47.     private static final String GGA_HEADER_KEY = "Ntrip-GGA";

  49.     /** Content type for GNSS data. */
  50.     private static final String GNSS_DATA_CONTENT_TYPE = "gnss/data";

  52.     /** Size of buffer for retrieving data. */
  53.     private static final int BUFFER_SIZE = 0x4000;

  55.     /** Frame preamble. */
  56.     private static final int PREAMBLE = 0xD3;

  58.     /** Frame preamble size. */
  59.     private static final int PREAMBLE_SIZE = 3;

  61.     /** Frame CRC size. */
  62.     private static final int CRC_SIZE = 3;

  64.     /** Generator polynomial for CRC. */
  65.     private static final int GENERATOR = 0x1864CFB;

  67.     /** High bit of the generator polynomial. */
  68.     private static final int HIGH = 0x1000000;

  70.     /** CRC 24Q lookup table. */
  71.     private static final int[] CRC_LOOKUP = new int[256];

  73.     static {

  75.         // set up lookup table
  76.         CRC_LOOKUP[0] = 0;
  77.         CRC_LOOKUP[1] = GENERATOR;

  79.         int h = GENERATOR;
  80.         for (int i = 2; i < 256; i <<= 1) {
  81.             h <<= 1;
  82.             if ((h & HIGH) != 0) {
  83.                 h ^= GENERATOR;
  84.             }
  85.             for (int j = 0; j < i; ++j) {
  86.                 CRC_LOOKUP[i + j] = CRC_LOOKUP[j] ^ h;
  87.             }
  88.         }

  90.     }

  92.     /** Associated NTRIP client. */
  93.     private final NtripClient client;

  95.     /** Mount point providing the stream. */
  96.     private final String mountPoint;

  98.     /** Messages type of the mount point. */
  99.     private final Type type;

  101.     /** Indicator for required NMEA. */
  102.     private final boolean nmeaRequired;

  104.     /** Indicator for ignoring unknown messages. */
  105.     private final boolean ignoreUnknownMessageTypes;

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

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

  113.     /** Max number of reconnections. */
  114.     private final int maxRetries;

  116.     /** Stop flag. */
  117.     private AtomicBoolean stop;

  119.     /** Circular buffer. */
  120.     private byte[] buffer;

  122.     /** Read index. */
  123.     private int readIndex;

  125.     /** Message end index. */
  126.     private int messageEndIndex;

  128.     /** Write index. */
  129.     private int writeIndex;

  131.     /** Observers for encoded messages. */
  132.     private final Map<Integer, List<MessageObserver>> observers;

  134.     /** Last available message for each type. */
  135.     private final Map<Integer, ParsedMessage> lastMessages;

  137.     /** Exception caught during monitoring. */
  138.     private final AtomicReference<OrekitException> exception;

  140.     /** Build a monitor for streaming data from a mount point.
  141.      * @param client associated NTRIP client
  142.      * @param mountPoint mount point providing the stream
  143.      * @param type messages type of the mount point
  144.      * @param requiresNMEA if true, the mount point requires a NMEA GGA sentence in the request
  145.      * @param ignoreUnknownMessageTypes if true, unknown messages types are silently ignored
  146.      * @param reconnectDelay delay before we reconnect after connection close
  147.      * @param reconnectDelayFactor factor by which reconnection delay is multiplied after each attempt
  148.      * @param maxRetries max number of reconnect a attempts without reading any data
  149.      */
  150.     public StreamMonitor(final NtripClient client,
  151.                          final String mountPoint, final Type type,
  152.                          final boolean requiresNMEA, final boolean ignoreUnknownMessageTypes,
  153.                          final double reconnectDelay, final double reconnectDelayFactor,
  154.                          final int maxRetries) {
  155.         this.client                    = client;
  156.         this.mountPoint                = mountPoint;
  157.         this.type                      = type;
  158.         this.nmeaRequired              = requiresNMEA;
  159.         this.ignoreUnknownMessageTypes = ignoreUnknownMessageTypes;
  160.         this.reconnectDelay            = reconnectDelay;
  161.         this.reconnectDelayFactor      = reconnectDelayFactor;
  162.         this.maxRetries                = maxRetries;
  163.         this.stop                      = new AtomicBoolean(false);
  164.         this.observers                 = new HashMap<>();
  165.         this.lastMessages              = new HashMap<>();
  166.         this.exception                 = new AtomicReference<OrekitException>(null);
  167.     }

  169.     /** Add an observer for encoded messages.
  170.      * <p>
  171.      * If messages of the specified type have already been retrieved from
  172.      * a stream, the observer will be immediately notified with the last
  173.      * message as a side effect of being added.
  174.      * </p>
  175.      * @param typeCode code for the message type (if set to 0, notification
  176.      * will be triggered regardless of message type)
  177.      * @param observer observer for this message type
  178.      */
  179.     public void addObserver(final int typeCode, final MessageObserver observer) {
  180.         synchronized (observers) {

  182.             // register the observer
  183.             List<MessageObserver> list = observers.get(typeCode);
  184.             if (list == null) {
  185.                 // create a new list the first time we register an observer for a message
  186.                 list =  new ArrayList<>();
  187.                 observers.put(typeCode, list);
  188.             }
  189.             list.add(observer);

  191.             // if we already have a message of the proper type
  192.             // immediately notify the new observer about it
  193.             final ParsedMessage last = lastMessages.get(typeCode);
  194.             if (last != null) {
  195.                 observer.messageAvailable(mountPoint, last);
  196.             }

  198.         }
  199.     }

  201.     /** Stop monitoring. */
  202.     public void stopMonitoring() {
  203.         stop.set(true);
  204.     }

  206.     /** Retrieve exception caught during monitoring.
  207.      * @return exception caught
  208.      */
  209.     public OrekitException getException() {
  210.         return exception.get();
  211.     }

  213.     /** {@inheritDoc} */
  214.     @Override
  215.     public void run() {

  217.         try {

  219.             final MessagesParser parser = type.getParser(extractUsedMessages());
  220.             int nbAttempts = 0;
  221.             double delay = reconnectDelay;
  222.             while (nbAttempts < maxRetries) {

  224.                 try {
  225.                     // prepare request
  226.                     final HttpURLConnection connection = client.connect(mountPoint);
  227.                     if (nmeaRequired) {
  228.                         if (client.getGGA() == null) {
  229.                             throw new OrekitException(OrekitMessages.STREAM_REQUIRES_NMEA_FIX, mountPoint);
  230.                         } else {
  231.                             // update NMEA GGA sentence in the extra headers for this mount point
  232.                             connection.setRequestProperty(GGA_HEADER_KEY, client.getGGA());
  233.                         }
  234.                     }

  236.                     // perform request
  237.                     final int responseCode = connection.getResponseCode();
  238.                     if (responseCode == HttpURLConnection.HTTP_UNAUTHORIZED) {
  239.                         throw new OrekitException(OrekitMessages.FAILED_AUTHENTICATION, mountPoint);
  240.                     } else if (responseCode != HttpURLConnection.HTTP_OK) {
  241.                         throw new OrekitException(OrekitMessages.CONNECTION_ERROR,
  242.                                                   connection.getURL().getHost(),
  243.                                                   connection.getResponseMessage());
  244.                     }

  246.                     // for this request, we MUST get GNSS data
  247.                     if (!GNSS_DATA_CONTENT_TYPE.equals(connection.getContentType())) {
  248.                         throw new OrekitException(OrekitMessages.UNEXPECTED_CONTENT_TYPE, connection.getContentType());
  249.                     }

  251.                     // data extraction loop
  252.                     resetCircularBuffer();
  253.                     try (InputStream is = connection.getInputStream()) {

  255.                         for (int r = fillUp(is); r >= 0; r = fillUp(is)) {

  257.                             // we have read something, reset reconnection attempts counters
  258.                             nbAttempts = 0;
  259.                             delay      = reconnectDelay;

  261.                             if (stop.get()) {
  262.                                 // stop monitoring immediately
  263.                                 // (returning closes the input stream automatically)
  264.                                 return;
  265.                             }

  267.                             while (bufferSize() >= 3) {
  268.                                 if (peekByte(0) != PREAMBLE) {
  269.                                     // we are out of synch with respect to frame structure
  270.                                     // drop the unknown byte
  271.                                     moveRead(1);
  272.                                 } else {
  273.                                     final int size = (peekByte(1) & 0x03) << 8 | peekByte(2);
  274.                                     if (bufferSize() >= PREAMBLE_SIZE + size + CRC_SIZE) {
  275.                                         // check CRC
  276.                                         final int crc = (peekByte(PREAMBLE_SIZE + size)     << 16) |
  277.                                                         (peekByte(PREAMBLE_SIZE + size + 1) <<  8) |
  278.                                                          peekByte(PREAMBLE_SIZE + size + 2);
  279.                                         if (crc == computeCRC(PREAMBLE_SIZE + size)) {
  280.                                             // we have a complete and consistent frame
  281.                                             // we can extract the message it contains
  282.                                             messageEndIndex = (readIndex + PREAMBLE_SIZE + size) % BUFFER_SIZE;
  283.                                             moveRead(PREAMBLE_SIZE);
  284.                                             start();
  285.                                             final ParsedMessage message = parser.parse(this, ignoreUnknownMessageTypes);
  286.                                             if (message != null) {
  287.                                                 storeAndNotify(message);
  288.                                             }
  289.                                             // jump to expected message end, in case the message was corrupted
  290.                                             // and parsing did not reach message end
  291.                                             readIndex = (messageEndIndex + CRC_SIZE) % BUFFER_SIZE;
  292.                                         } else {
  293.                                             // CRC is not consistent, we are probably not really synched
  294.                                             // and the preamble byte was just a random byte
  295.                                             // we drop this single byte and continue looking for sync
  296.                                             moveRead(1);
  297.                                         }
  298.                                     } else {
  299.                                         // the frame is not complete, we need more data
  300.                                         break;
  301.                                     }
  302.                                 }
  303.                             }

  305.                         }

  307.                     }
  308.                 } catch (SocketTimeoutException ste) {
  309.                     // ignore exception, it will be handled by reconnection attempt below
  310.                 } catch (IOException | URISyntaxException e) {
  311.                     throw new OrekitException(e, OrekitMessages.CANNOT_PARSE_GNSS_DATA, client.getHost());
  312.                 }

  314.                 // manage reconnection
  315.                 try {
  316.                     Thread.sleep((int) FastMath.rint(delay * 1000));
  317.                 } catch (InterruptedException ie) {
  318.                     // Restore interrupted state...
  319.                     Thread.currentThread().interrupt();
  320.                 }
  321.                 ++nbAttempts;
  322.                 delay *= reconnectDelayFactor;

  324.             }

  326.         } catch (OrekitException oe) {
  327.             // store the exception so it can be retrieved by Ntrip client
  328.             exception.set(oe);
  329.         }

  331.     }

  333.     /** Store a parsed encoded message and notify observers.
  334.      * @param message parsed message
  335.      */
  336.     private void storeAndNotify(final ParsedMessage message) {
  337.         synchronized (observers) {

  339.             for (int typeCode : Arrays.asList(0, message.getTypeCode())) {

  341.                 // store message
  342.                 lastMessages.put(typeCode, message);

  344.                 // notify observers
  345.                 final List<MessageObserver> list = observers.get(typeCode);
  346.                 if (list != null) {
  347.                     for (final MessageObserver observer : list) {
  348.                         // notify observer
  349.                         observer.messageAvailable(mountPoint, message);
  350.                     }
  351.                 }

  353.             }

  355.         }
  356.     }

  358.     /** Reset the circular buffer.
  359.      */
  360.     private void resetCircularBuffer() {
  361.         buffer     = new byte[BUFFER_SIZE];
  362.         readIndex  = 0;
  363.         writeIndex = 0;
  364.     }

  366.     /** Extract data from input stream.
  367.      * @param is input stream to extract data from
  368.      * @return number of byes read or -1
  369.      * @throws IOException if data cannot be extracted properly
  370.      */
  371.     private int fillUp(final InputStream is) throws IOException {
  372.         final int max = bufferMaxWrite();
  373.         if (max == 0) {
  374.             // this should never happen
  375.             // the buffer is large enough for almost 16 encoded messages, including wrapping frame
  376.             throw new OrekitInternalError(null);
  377.         }
  378.         final int r = is.read(buffer, writeIndex, max);
  379.         if (r >= 0) {
  380.             writeIndex = (writeIndex + r) % BUFFER_SIZE;
  381.         }
  382.         return r;
  383.     }

  385.     /** {@inheritDoc} */
  386.     @Override
  387.     protected int fetchByte() {
  388.         if (readIndex == messageEndIndex || readIndex == writeIndex) {
  389.             return -1;
  390.         }

  392.         final int ret = buffer[readIndex] & 0xFF;
  393.         moveRead(1);
  394.         return ret;
  395.     }

  397.     /** Get the number of bytes currently in the buffer.
  398.      * @return number of bytes currently in the buffer
  399.      */
  400.     private int bufferSize() {
  401.         final int n = writeIndex - readIndex;
  402.         return n >= 0 ? n : BUFFER_SIZE + n;
  403.     }

  405.     /** Peek a buffer byte without moving read pointer.
  406.      * @param offset offset counted from read pointer
  407.      * @return value of the byte at given offset
  408.      */
  409.     private int peekByte(final int offset) {
  410.         return buffer[(readIndex + offset) % BUFFER_SIZE] & 0xFF;
  411.     }

  413.     /** Move read pointer.
  414.      * @param n number of bytes to move read pointer
  415.      */
  416.     private void moveRead(final int n) {
  417.         readIndex = (readIndex + n) % BUFFER_SIZE;
  418.     }

  420.     /** Get the number of bytes that can be added to the buffer without wrapping around.
  421.      * @return number of bytes that can be added
  422.      */
  423.     private int bufferMaxWrite() {
  424.         if (writeIndex >= readIndex) {
  425.             return (readIndex == 0 ? BUFFER_SIZE - 1 : BUFFER_SIZE) - writeIndex;
  426.         } else {
  427.             return readIndex - writeIndex - 1;
  428.         }
  429.     }

  431.     /** Compute QualCom CRC.
  432.      * @param length length of the byte stream
  433.      * @return QualCom CRC
  434.      */
  435.     private int computeCRC(final int length) {
  436.         int crc = 0;
  437.         for (int i = 0; i < length; ++i) {
  438.             crc = ((crc << 8) ^ CRC_LOOKUP[peekByte(i) ^ (crc >>> 16)]) & (HIGH - 1);
  439.         }
  440.         return crc;
  441.     }

  443.     /** Extract the used messages.
  444.      * @return the extracted messages
  445.      */
  446.     private List<Integer> extractUsedMessages() {
  447.         synchronized (observers) {

  449.             // List of needed messages
  450.             final List<Integer> messages = new ArrayList<>();

  452.             // Loop on observers entries
  453.             for (Map.Entry<Integer, List<MessageObserver>> entry : observers.entrySet()) {
  454.                 // Extract message type code
  455.                 final int typeCode = entry.getKey();
  456.                 // Add to the list
  457.                 messages.add(typeCode);
  458.             }

  460.             return messages;
  461.         }
  462.     }

  464. }
Created with JaCoCo 0.8.11.202310140853