- included message throttling capability and auto throttling capability to address processor bog-down during heavy XML operations

Author: jvinall

SweLib/build.gradle

@@ -9,8 +9,8 @@ android {
     defaultConfig {
         minSdkVersion 21
         targetSdkVersion 28
-        versionCode 7
-        versionName '1.0.7'
+        versionCode 8
+        versionName '1.0.8'
     }
     buildTypes {
         release {

SweLib/src/main/java/org/sofwerx/ogc/sos/OperationGetResults.java

@@ -39,7 +39,9 @@ public ArrayList<Pair<String,String>> getPairs() {
         pairs.add(new Pair("request","GetResult"));
         pairs.add(new Pair("offering",sensor.getAssignedOffering()));
         pairs.add(new Pair("observedProperty",sensor.getFirstObservableProperty()));
+        //TODO pairs.add(new Pair("temporalFilter","phenomenonTime,now")); //this sisnt working for some reason
         pairs.add(new Pair("responseFormat","application/json"));
+
         return pairs;
     }
 

SweLib/src/main/java/org/sofwerx/ogc/sos/SosIpcTransceiver.java

@@ -45,17 +45,42 @@ public class SosIpcTransceiver extends BroadcastReceiver {
     public static final String ACTION_SQAN_BROADCAST = "org.sofwerx.sqan.pkt";
     private final static String SQAN_PACKET_BYTES = "bytes";
     private final static String SQAN_PACKET_CHANNEL = "channel";
+    final static long DEFAULT_OUTGOING_THROTTLE_RATE = 1000l * 5l;
+    final static long DEFAULT_INCOMING_THROTTLE_RATE = 1000l * 10l;
     private static boolean enableSqAN = true;
     private static String channel = SosService.DEFAULT_SWE_CHANNEL;
     private SosMessageListener listener;
 
+    //Since XML is expensive to marshall/unmarshall, include throttling to ignore messages that come in or go out too fast
+    private static long throttleRate = -1l;
+    private static long nextAvailableIntake = Long.MIN_VALUE;
+
     public SosIpcTransceiver(SosMessageListener listener) {
         this.listener = listener;
     }
 
     public static void setChannel(String channel) { SosIpcTransceiver.channel = channel; }
     public static void setEnableSqAN(boolean enable) { SosIpcTransceiver.enableSqAN = enable; }
 
+    /**
+     * Sets a throttle rate (i.e. a min amo8nt of time between messages in ms); all messages
+     * received in excess of the throttle rate will be dropped (done to prevent bogging
+     * down the processor in XML marshallig/unmarshalling operations).
+     * @param rate rate in ms (or -1l if no throttling is needed)
+     */
+    public static void setThrottleRate(long rate) {
+        if (throttleRate != rate) {
+            throttleRate = rate;
+            if (rate > 0l)
+                Log.d(TAG, "Setting throttle interval to " + Long.toString(rate) + "ms");
+            else {
+                Log.d(TAG, "Removing throttle");
+            }
+        }
+    }
+
+    public static void clearThrottle() { setThrottleRate(-1l); }
+
     @Override
     public void onReceive(Context context, Intent intent) {
         if ((context != null) && (intent != null)) {
@@ -98,22 +123,27 @@ public void onMessageReceived(final Context context,final String source, final S
             Log.e(TAG, "Null operation received from SOS broadcast IPC");
             return;
         }
-        new Thread(() -> {
-            try {
-                DocumentBuilderFactory builderFactory = DocumentBuilderFactory.newInstance();
-                DocumentBuilder docBuilder = builderFactory.newDocumentBuilder();
-                Document doc = docBuilder.parse(new InputSource(new ByteArrayInputStream(input.getBytes("utf-8"))));
-                if (doc != null) {
-                    AbstractSosOperation operation = AbstractSosOperation.newFromXML(doc);
-                    if (operation != null) {
-                        if (listener != null)
-                            listener.onSosOperationReceived(operation);
+        if ((throttleRate <= 0l) || (System.currentTimeMillis() > nextAvailableIntake)) {
+            if (throttleRate > 0l)
+                nextAvailableIntake = System.currentTimeMillis() + throttleRate;
+            new Thread(() -> {
+                try {
+                    DocumentBuilderFactory builderFactory = DocumentBuilderFactory.newInstance();
+                    DocumentBuilder docBuilder = builderFactory.newDocumentBuilder();
+                    Document doc = docBuilder.parse(new InputSource(new ByteArrayInputStream(input.getBytes("utf-8"))));
+                    if (doc != null) {
+                        AbstractSosOperation operation = AbstractSosOperation.newFromXML(doc);
+                        if (operation != null) {
+                            if (listener != null)
+                                listener.onSosOperationReceived(operation);
+                        }
                     }
+                } catch (ParserConfigurationException | IOException | SAXException e) {
+                    Log.e(TAG, "SOS IPC broadcast was not XML: " + input);
                 }
-            } catch (ParserConfigurationException | IOException | SAXException e) {
-                Log.e(TAG,"SOS IPC broadcast was not XML: "+input);
-            }
-        }).start();
+            }).start();
+        } else
+            Log.d(TAG,"Dropping message from "+source+" due to flooding: "+input);
     }
 
     /**
@@ -122,25 +152,30 @@ public void onMessageReceived(final Context context,final String source, final S
      * @param operation
      */
     public void broadcast(final Context context, final AbstractSosOperation operation) throws SosException {
-        if (operation != null) {
-            if (!operation.isValid()) {
-                throw new SosException(operation.getClass().getSimpleName()+" does not have all required information");
-            }
-            new Thread(() -> {
-                Document doc = null;
-                try {
-                    doc = operation.toXML();
-                } catch (ParserConfigurationException e) {
-                    //throw new SosException("Unable to create document: " + e.getMessage());
-                }
-                try {
-                    if (doc != null)
-                        broadcast(context, toString(doc));
-                } catch (Exception ex) {
-                    //throw new SosException("Unable to convert XML document to string: " + ex.getMessage());
+        if ((throttleRate <= 0l) || (System.currentTimeMillis() > nextAvailableIntake)) {
+            if (throttleRate > 0l)
+                nextAvailableIntake = System.currentTimeMillis() + throttleRate;
+            if (operation != null) {
+                if (!operation.isValid()) {
+                    throw new SosException(operation.getClass().getSimpleName() + " does not have all required information");
                 }
-            }).start();
-        }
+                new Thread(() -> {
+                    Document doc = null;
+                    try {
+                        doc = operation.toXML();
+                    } catch (ParserConfigurationException e) {
+                        //throw new SosException("Unable to create document: " + e.getMessage());
+                    }
+                    try {
+                        if (doc != null)
+                            broadcast(context, toString(doc));
+                    } catch (Exception ex) {
+                        //throw new SosException("Unable to convert XML document to string: " + ex.getMessage());
+                    }
+                }).start();
+            }
+        } else
+            Log.d(TAG,operation.getClass().getSimpleName()+" operation received but ignored since the current throttle rate of "+Long.toString(throttleRate)+"ms is being exceeded");
     }
 
     public final static String toString(Document doc) throws TransformerException {

SweLib/src/main/java/org/sofwerx/ogc/sos/SosService.java

@@ -40,6 +40,9 @@ public class SosService implements SosMessageListener {
     private boolean sensorMode = true;
     private AtomicBoolean shouldPollServer = new AtomicBoolean(false);
     private long pollInterval = 1000l * 15l; //default server polling interval
+    private boolean autoThrottle = false;
+    private long outgoingThrottleRate = SosIpcTransceiver.DEFAULT_OUTGOING_THROTTLE_RATE;
+    private long incomingThrottleRate = SosIpcTransceiver.DEFAULT_INCOMING_THROTTLE_RATE;
 
     /**
      * Creates a new SosService
@@ -274,13 +277,17 @@ public void registerSensor() {
             handler.post(() -> {
                 if (sosSensor.getAssignedProcedure() == null) {
                     if (sosSensor.isReadyToRegisterSensor()) {
+                        if (autoThrottle)
+                            SosIpcTransceiver.clearThrottle();
                         Log.d(SosIpcTransceiver.TAG,"Sensor has all required info to register; contacting server...");
                         OperationInsertSensor operation = new OperationInsertSensor(sosSensor);
                         broadcast(operation);
                     } else
                         Log.w(SosIpcTransceiver.TAG, "sosSensor does not yet have enough information to register with the SOS server");
                 } else if (sosSensor.getAssignedTemplate() == null) {
                     if (sosSensor.isReadyToRegisterResultTemplate()) {
+                        if (autoThrottle)
+                            SosIpcTransceiver.clearThrottle();
                         OperationInsertResultTemplate operation = new OperationInsertResultTemplate(sosSensor);
                         broadcast(operation);
                     } else
@@ -339,6 +346,8 @@ public void onSosOperationReceived(AbstractSosOperation operation) {
                     Log.i(SosIpcTransceiver.TAG,"InsertSensorResponse received, but it was for sensor "+response.getAssignedProcedure());
             }
         } else if (operation instanceof OperationInsertResultTemplateResponse) {
+            if (autoThrottle)
+                SosIpcTransceiver.setThrottleRate(outgoingThrottleRate);
             if (sosSensor != null) {
                 OperationInsertResultTemplateResponse response = (OperationInsertResultTemplateResponse) operation;
                 if ((response.getAcceptedTemplate() != null) && (sosSensor.getAssignedProcedure() != null)
@@ -349,8 +358,14 @@ public void onSosOperationReceived(AbstractSosOperation operation) {
                     if (listener != null)
                         listener.onSosConfigurationSuccess();
                 } else
-                    Log.i(SosIpcTransceiver.TAG,"InsertResultTemplateResponse received, but it was for template "+response.getAcceptedTemplate());
+                    Log.i(SosIpcTransceiver.TAG, "InsertResultTemplateResponse received, but it was for template " + response.getAcceptedTemplate());
             }
+        } else if (operation instanceof OperationInsertResult) {
+            if (autoThrottle)
+                SosIpcTransceiver.setThrottleRate(incomingThrottleRate);
+        } else if (operation instanceof OperationInsertResultResponse) {
+            if (autoThrottle)
+                SosIpcTransceiver.setThrottleRate(outgoingThrottleRate);
         }
         if (listener != null)
             listener.onSosOperationReceived(operation);
@@ -402,4 +417,48 @@ public boolean isSensorMode() {
      */
     public void setSensorMode(boolean sensorMode) { this.sensorMode = sensorMode; }
     public void setSensorMode() { setSensorMode(true); }
+
+    /**
+     * Is the service set to automatically throttle incoming messages to offset
+     * cost on flooding XML unmarshalling.
+     * @return true == will automatically throttle
+     */
+    public boolean isAutoThrottle() { return autoThrottle; }
+
+    /**
+     * Sets the service to automatically throttle incoming messages (usually done as
+     * a way to prevent flooding the processor with a bunch of costly XML unmarshalling)
+     * @param autoThrottle true == the service will automatically throttle
+     */
+    public void setAutoThrottle(boolean autoThrottle) { this.autoThrottle = autoThrottle; }
+
+    /**
+     * Gets the set throttle rate on outgoing messages
+     * @return interval between messages in ms
+     */
+    public long getOutgoingThrottleRate() { return outgoingThrottleRate; }
+
+    /**
+     * Sets the set throttle rate on outgoing messages
+     * @param outgoingThrottleRate interval between messages in ms
+     */
+    public void setOutgoingThrottleRate(long outgoingThrottleRate) {
+        if (outgoingThrottleRate > 0l)
+            this.outgoingThrottleRate = outgoingThrottleRate;
+    }
+
+    /**
+     * Gets the set throttle rate on incoming messages
+     * @return interval between messages in ms
+     */
+    public long getIncomingThrottleRate() { return incomingThrottleRate; }
+
+    /**
+     * Sets the set throttle rate on incoming messages
+     * @param incomingThrottleRate interval between messages in ms
+     */
+    public void setIncomingThrottleRate(long incomingThrottleRate) {
+        if (incomingThrottleRate > 0l)
+            this.incomingThrottleRate = incomingThrottleRate;
+    }
 }