Merge pull request apache#77

This closes apache#77

analytics/sensors/src/main/java/quarks/analytics/sensors/Filters.java

@@ -99,7 +99,7 @@ public static <T, V> TStream<T> deadband(TStream<T> stream, Function<T, V> value
      * <LI>The first tuple.</LI>
      * </UL>
      * <BR>
-     * <img src="doc-files/deadband.png" alt="Deadband example"/>
+     * <img src="doc-files/deadband.png" alt="Deadband example">
      * </P>
      * 
      * @param <T> Tuple type.

analytics/sensors/src/main/java/quarks/analytics/sensors/Range.java

@@ -159,7 +159,7 @@ private Range(T lowerEndpoint, BoundType lbt, T upperEndpoint, BoundType ubt) {
     }
 
     /**
-     * Create a new Range<T>
+     * Create a new Range&lt;T&gt;
      * <p>
      * See {@link Ranges} for a collection of convenience constructors.
      * 
@@ -257,13 +257,13 @@ public BoundType upperBoundType() {
      * can be used in cases where it isn't sufficient.
      * E.g., for unsigned byte comparisons
      * <pre>
-     * Comparator<Byte> unsignedByteComparator = new Comparator<Byte>() {
+     * Comparator&lt;Byte&gt; unsignedByteComparator = new Comparator&lt;Byte&gt;() {
      *     public int compare(Byte b1, Byte b2) {
      *         return Integer.compareUnsigned(Byte.toUnsignedInt(b1), Byte.toUnsignedInt(b2));
      *     }
      *     public boolean equals(Object o2) { return o2==this; }
      *     };
-     * Range<Byte> unsignedByteRange = Ranges.valueOfByte("[0..255]");
+     * Range&lt;Byte&gt; unsignedByteRange = Ranges.valueOfByte("[0..255]");
      * unsignedByteRange.contains(byteValue, unsignedByteComparator);
      * </pre>
      * 
@@ -410,7 +410,7 @@ public static <T extends Comparable<?>> Range<T> valueOf(String toStringValue, F
      * See toStringUnsigned().
      * <p>
      * No special processing is performed to escape/encode a "." present
-     * in an endpoint.toString() value.  Hence Range<T>.toString() for
+     * in an endpoint.toString() value.  Hence Range&lt;T&gt;.toString() for
      * a {@code T} of {@code String} (of value "." or with embedded ".."),
      * or some other non-numeric type may yield values that are not amenable
      * to parsing by {@link #valueOf(String, Function)}.

analytics/sensors/src/main/java/quarks/analytics/sensors/Ranges.java

@@ -96,7 +96,7 @@ public static  <T extends Comparable<?>> Range<T> singleton(T endpoint) {
     }
 
     /**
-     * Create a Range from a Range<Integer>.toString() value. 
+     * Create a Range from a Range&lt;Integer&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -107,7 +107,7 @@ public static Range<Integer> valueOfInteger(String str) {
     }
 
     /**
-     * Create a Range from a Range<Short>.toString() value. 
+     * Create a Range from a Range&lt;Short&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -118,7 +118,7 @@ public static Range<Short> valueOfShort(String str) {
     }
 
     /**
-     * Create a Range from a Range<Byte>.toString() value. 
+     * Create a Range from a Range&lt;Byte&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -129,7 +129,7 @@ public static Range<Byte> valueOfByte(String str) {
     }
 
     /**
-     * Create a Range from a Range<Long>.toString() value. 
+     * Create a Range from a Range&lt;Long&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -140,7 +140,7 @@ public static Range<Long> valueOfLong(String str) {
     }
 
     /**
-     * Create a Range from a Range<Float>.toString() value. 
+     * Create a Range from a Range&lt;Float&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -151,7 +151,7 @@ public static Range<Float> valueOfFloat(String str) {
     }
 
     /**
-     * Create a Range from a Range<Double>.toString() value. 
+     * Create a Range from a Range&lt;Double&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -162,7 +162,7 @@ public static Range<Double> valueOfDouble(String str) {
     }
 
     /**
-     * Create a Range from a Range<BigInteger>.toString() value. 
+     * Create a Range from a Range&lt;BigInteger&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -173,7 +173,7 @@ public static Range<BigInteger> valueOfBigInteger(String str) {
     }
 
     /**
-     * Create a Range from a Range<BigDecimal>.toString() value. 
+     * Create a Range from a Range&lt;BigDecimal&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert
@@ -184,7 +184,7 @@ public static Range<BigDecimal> valueOfBigDecimal(String str) {
     }
 
     /**
-     * Create a Range from a Range<String>.toString() value. 
+     * Create a Range from a Range&lt;String&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if str includes a String
@@ -195,7 +195,7 @@ public static Range<String> valueOfString(String str) {
     }
 
     /**
-     * Create a Range from a Range<Character>.toString() value. 
+     * Create a Range from a Range&lt;Character&gt;.toString() value.
      * @param str the String
      * @return the Range
      * @throws IllegalArgumentException if unable to parse or convert

api/graph/src/main/java/quarks/graph/Connector.java

@@ -43,13 +43,13 @@ Licensed to the Apache Software Foundation (ASF) under one
  * </UL>
  * For example with peeks {@code P1,P2,P3} added in that order and connections
  * {@code C1,C2} added, the graph will be logically:
- * 
+ *
  * <pre>
- * <code>
+ * {@code
  *                      -->C1
  * port-->P1-->P2-->P3--|
  *                      -->C2
- * </code>
+ * }
  * </pre>
  * 
  * A tuple {@code t} submitted by the port will be peeked at by {@code P1}, then

api/topology/src/main/java/quarks/topology/TStream.java

@@ -65,8 +65,8 @@ public interface TStream<T> extends TopologyElement {
      * 
      * <pre>
      * <code>
-     * TStream&lt;String> s = ...
-     * TStream&lt;String> filtered = s.filter(t -> !t.isEmpty());
+     * TStream&lt;String&gt; s = ...
+     * TStream&lt;String&gt; filtered = s.filter(t -&gt; !t.isEmpty());
      *             
      * </code>
      * </pre>
@@ -94,12 +94,12 @@ public interface TStream<T> extends TopologyElement {
      * <pre>
      * <code>
      * // Using lambda expression
-     * TStream&lt;String> strings = ...
-     * TStream&lt;Double> doubles = strings.map(v -> Double.valueOf(v));
+     * TStream&lt;String&gt; strings = ...
+     * TStream&lt;Double&gt; doubles = strings.map(v -&gt; Double.valueOf(v));
      * 
      * // Using method reference
-     * TStream&lt;String> strings = ...
-     * TStream&lt;Double> doubles = strings.map(Double::valueOf);
+     * TStream&lt;String&gt; strings = ...
+     * TStream&lt;Double&gt; doubles = strings.map(Double::valueOf);
      * 
      * </code>
      * </pre>
@@ -130,9 +130,9 @@ public interface TStream<T> extends TopologyElement {
      * 
      * <pre>
      * <code>
-     * TStream&lt;String> lines = ...
-     * TStream&lt;String> words = lines.flatMap(
-     *                     line -> Arrays.asList(line.split(" ")));
+     * TStream&lt;String&gt; lines = ...
+     * TStream&lt;String&gt; words = lines.flatMap(
+     *                     line -&gt; Arrays.asList(line.split(" ")));
      *             
      * </code>
      * </pre>
@@ -155,7 +155,7 @@ public interface TStream<T> extends TopologyElement {
      * called. The return value {@code r} determines the destination stream:
      * 
      * <pre>
-     * if r < 0 the tuple is discarded
+     * if r &lt; 0 the tuple is discarded
      * else it is sent to the stream at position (r % n) in the returned array.
      * </pre>
      * </P>
@@ -174,19 +174,19 @@ public interface TStream<T> extends TopologyElement {
      * destination stream. For example, these are logically equivalent:
      * 
      * <pre>
-     * List&lt;TStream&lt;String>> streams = stream.split(2, tuple -> tuple.length());
+     * List&lt;TStream&lt;String&gt;&gt; streams = stream.split(2, tuple -&gt; tuple.length());
      * 
-     * TStream&lt;String> stream0 = stream.filter(tuple -> (tuple.length() % 2) == 0);
-     * TStream&lt;String> stream1 = stream.filter(tuple -> (tuple.length() % 2) == 1);
+     * TStream&lt;String&gt; stream0 = stream.filter(tuple -&gt; (tuple.length() % 2) == 0);
+     * TStream&lt;String&gt; stream1 = stream.filter(tuple -&gt; (tuple.length() % 2) == 1);
      * </pre>
      * </P>
      * <P>
      * Example of splitting a stream of log records by their level attribute:
      * 
      * <pre>
      * <code>
-     * TStream&lt;LogRecord> lrs = ...
-     * List&lt;&lt;TStream&lt;LogRecord>> splits = lrr.split(3, lr -> {
+     * TStream&lt;LogRecord&gt; lrs = ...
+     * List&lt;&lt;TStream&lt;LogRecord&gt;&gt; splits = lrr.split(3, lr -&gt; {
             if (SEVERE.equals(lr.getLevel()))
                 return 0;
             else if (WARNING.equals(lr.getLevel()))
@@ -220,7 +220,7 @@ else if (WARNING.equals(lr.getLevel()))
      *            enum data to split
      * @param splitter
      *            the splitter function
-     * @return EnumMap<E,TStream<T>>
+     * @return EnumMap&lt;E,TStream&lt;T&gt;&gt;
      * @throws IllegalArgumentException
      * if {@code enumclass.size <= 0}
      */
@@ -252,8 +252,8 @@ else if (WARNING.equals(lr.getLevel()))
      * 
      * <pre>
      * <code>
-     * TStream&lt;String> values = ...
-     * values.sink(t -> System.out.println(tuple));
+     * TStream&lt;String&gt; values = ...
+     * values.sink(t -&gt; System.out.println(tuple));
      * </code>
      * </pre>
      * 
@@ -304,16 +304,16 @@ else if (WARNING.equals(lr.getLevel()))
      * 
      * <pre>
      * <code>
-     * TStream&lt;String> strings = ...
-     * TStream&lt;String> modifiedStrings = strings.modify(t -> t.concat("extra"));
+     * TStream&lt;String&gt; strings = ...
+     * TStream&lt;String&gt; modifiedStrings = strings.modify(t -&gt; t.concat("extra"));
      * </code>
      * </pre>
      * 
      * </P>
      * <P>
      * This method is equivalent to
      * {@code map(Function<T,T> modifier}).
-     * </P
+     * </P>
      * 
      * @param modifier
      *            Modifier logic to be executed against each tuple.

apps/runtime/src/main/java/quarks/apps/runtime/JobMonitorApp.java

@@ -44,13 +44,14 @@ Licensed to the Apache Software Foundation (ASF) under one
 
 /**
  * Job monitoring application.
- * <p>
+ * <P>
  * The application listens on JobRegistry events and resubmits jobs for which 
  * an event has been emitted because the job is unhealthy. The monitored 
  * applications must be registered with an {@code ApplicationService} 
  * prior to submission, otherwise the monitor application cannot restart 
- * them.</p>
- * <p> 
+ * them.
+ * </P>
+ * <P>
  * The monitoring application must be submitted within a context which 
  * provides the following services:
  * <ul>
@@ -61,7 +62,7 @@ Licensed to the Apache Software Foundation (ASF) under one
  * restarting failed applications.</li>
  * <li>JobRegistryService - generates job monitoring events. </li>
  * </ul>
- * </p>
+ * </P>
  */
 public class JobMonitorApp {
     /**
@@ -138,7 +139,7 @@ public static void submitApplication(String applicationName, ControlService cont
     /**
      * Declares the following topology:
      * <pre>
-     * JobEvents source --> Filter (health == unhealthy) --> Restart application
+     * JobEvents source --&gt; Filter (health == unhealthy) --&gt; Restart application
      * </pre>
      * 
      * @param name the topology name

connectors/common/src/main/java/quarks/connectors/runtime/Connector.java

@@ -54,7 +54,8 @@ Licensed to the Apache Software Foundation (ASF) under one
  * <ul>
  * <li>[abstract] ConnectorControl { getState(), connect(), disconnect(), ...}.</li>
  * <li>MqttConnectorControl extends ConnectorControl.  Adds subscribe(), unsubscribe().</li>
- * <li>Add Consumer<MqttConnectorControl> to MqttConfig. Runtime calls it supplying a connector control object.  By doing a disconnect()/connect(), the Supplier<MqttConfig> will be called to as part of reconnect -- hence using updated values if any.</li>
+ * <li>Add Consumer&lt;MqttConnectorControl&gt; to MqttConfig. Runtime calls it supplying a connector control object.
+ * By doing a disconnect()/connect(), the Supplier&lt;MqttConfig&gt; will be called to as part of reconnect -- hence using updated values if any.</li>
  * </ul>
  * <p>
  * Sub-classes are responsible for implementing a small number

connectors/file/src/main/java/quarks/connectors/file/FileWriterPolicy.java

@@ -54,14 +54,14 @@ Licensed to the Apache Software Foundation (ASF) under one
  * <p>
  * Sample use:
  * <pre>
- * FileWriterPolicy<String> policy = new FileWriterPolicy(
+ * FileWriterPolicy&lt;String&gt; policy = new FileWriterPolicy(
  *     FileWriterFlushConfig.newImplicitConfig(),
  *     FileWriterCycleConfig.newCountBasedConfig(1000),
  *     FileWriterRetentionConfig.newCountBasedConfig(10));
  * String basePathname = "/some/directory/and_base_name";
  * 
- * TStream<String> streamToWrite = ...
- * FileStreams.textFileWriter(streamToWrite, () -> basePathname, () -> policy)
+ * TStream&lt;String&gt; streamToWrite = ...
+ * FileStreams.textFileWriter(streamToWrite, () -&gt; basePathname, () -&gt; policy)
  * </pre>
  * 
  * @param <T> stream tuple type

connectors/file/src/main/java/quarks/connectors/file/runtime/IFileWriterPolicy.java

@@ -42,8 +42,8 @@ Licensed to the Apache Software Foundation (ASF) under one
  * <p>
  * A file writer uses a {@code IFileWriterPolicy} in the following manner:
  * <pre>
- * IFileWriterPolicy<T> policy = some policy implementation
- * policy.initialize(basePathname, () -> myFlushFn(), () -> myCycleFn());
+ * IFileWriterPolicy&lt;T&gt; policy = some policy implementation
+ * policy.initialize(basePathname, () -&gt; myFlushFn(), () -&gt; myCycleFn());
  * Path activeFilePath = null;
  * for each tuple {
  *   if (activePathFile == null) {

connectors/http/src/main/java/quarks/connectors/http/HttpClients.java

@@ -34,7 +34,7 @@ Licensed to the Apache Software Foundation (ASF) under one
  * This methods are called at runtime to create
  * HTTP clients for {@link HttpStreams}. They are
  * passed into methods such as
- * @link {@link HttpStreams#requests(quarks.topology.TStream, Supplier, quarks.function.Function, quarks.function.Function, quarks.function.BiFunction)}
+ * {@link HttpStreams#requests(quarks.topology.TStream, Supplier, quarks.function.Function, quarks.function.Function, quarks.function.BiFunction)}
  * as functions, for example:
  * <UL style="list-style-type:none">
  * <LI>{@code () -> HttpClients::noAuthentication } // using a method reference</LI>

connectors/http/src/main/java/quarks/connectors/http/HttpResponders.java

@@ -82,11 +82,12 @@ public static <T> BiFunction<T, CloseableHttpResponse, T> inputOn(Integer... cod
      * For each HTTP response a JSON object is produced that contains:
      * <UL>
      * <LI> {@code request} - the original input tuple that lead to the request  </LI>
-     * <LI> {@code response} - JSON object containing information about the response</LI>
+     * <LI> {@code response} - JSON object containing information about the response
      * <UL>
      *    <LI> {@code status} - Status code for the response as an integer</LI>
      *    <LI> {@code entity} - JSON response entity if one exists </LI>
      * </UL>
+     * </LI>
      * </UL>
      * 
      * @return Function that will process the {@code application/json} responses.

connectors/iot/src/main/java/quarks/connectors/iot/IotDevice.java

@@ -115,11 +115,12 @@ TSink<JsonObject> events(TStream<JsonObject> stream, Function<JsonObject, String
      * <LI>{@link #CMD_ID command} - Command identifier as a String</LI>
      * <LI>{@link #CMD_TS tsms} - Timestamp of the command in milliseconds since the 1970/1/1 epoch.</LI>
      * <LI>{@link #CMD_FORMAT format} - Format of the command as a String</LI>
-     * <LI>{@link #CMD_PAYLOAD payload} - Payload of the command</LI>
+     * <LI>{@link #CMD_PAYLOAD payload} - Payload of the command
      * <UL>
      * <LI>If {@code format} is {@code json} then {@code payload} is JSON</LI>
-     * <LI>Otherwise {@code payload} is String
+     * <LI>Otherwise {@code payload} is String</LI>
      * </UL>
+     * </LI>
      * </UL>
      * 
      * 

connectors/iot/src/main/java/quarks/connectors/iot/package-info.java

@@ -48,8 +48,9 @@ Licensed to the Apache Software Foundation (ASF) under one
  * </UL>
  * Device commands can be used to perform any action on the device including displaying information,
  * controlling the device (e.g. reduce maximum engine revolutions), controlling the Quarks application, etc.
- * </LI>
+ * <BR>
  * The format for the payload is typically JSON, though other formats may be used.
+ * </LI>
  * </UL>
  * </P>
  * <P>