You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by dl...@apache.org on 2016/06/13 14:22:27 UTC
[3/6] incubator-quarks git commit: fix javadoc warnings and errors
when Xdoclint:none is disabled
fix javadoc warnings and errors when Xdoclint:none is disabled
- lots of "unexpected end tag </P>" errors often due to it following a
table or list (which automatically ends a preceeding <P>)
- lots of missing generic type param doc, @return and @param
- also with enabling Eclipse > Java > Compiler > Javadoc > Missing
Javadoc Tags
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/d7840011
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/d7840011
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/d7840011
Branch: refs/heads/master
Commit: d78400110fa0cc0f8904c3f12205ce778eb4b9c4
Parents: 7924cc6
Author: Dale LaBossiere <dl...@us.ibm.com>
Authored: Fri Jun 10 10:31:15 2016 -0400
Committer: Dale LaBossiere <dl...@us.ibm.com>
Committed: Fri Jun 10 10:31:15 2016 -0400
----------------------------------------------------------------------
.../java/quarks/analytics/sensors/Deadtime.java | 2 +-
.../java/quarks/analytics/sensors/Filters.java | 2 +-
.../java/quarks/analytics/sensors/Range.java | 6 ++-
.../java/quarks/analytics/sensors/Ranges.java | 40 ++++++++++++++++++++
.../src/main/java/quarks/execution/Configs.java | 3 +-
.../quarks/execution/mbeans/PeriodMXBean.java | 3 ++
.../execution/services/ControlService.java | 8 +++-
.../quarks/execution/services/Controls.java | 2 +-
.../execution/services/JobRegistryService.java | 3 +-
.../execution/services/RuntimeServices.java | 4 +-
.../execution/services/ServiceContainer.java | 3 ++
.../main/java/quarks/function/BiConsumer.java | 2 +
.../main/java/quarks/function/Functions.java | 18 ++++++++-
.../java/quarks/function/WrappedFunction.java | 2 +
.../src/main/java/quarks/graph/Connector.java | 5 ++-
api/graph/src/main/java/quarks/graph/Graph.java | 11 +++++-
api/oplet/src/main/java/quarks/oplet/Oplet.java | 4 +-
.../main/java/quarks/oplet/OpletContext.java | 4 +-
.../src/main/java/quarks/oplet/core/FanIn.java | 7 +++-
.../src/main/java/quarks/oplet/core/Union.java | 2 +
.../quarks/oplet/plumbing/PressureReliever.java | 4 +-
.../src/main/java/quarks/topology/TSink.java | 1 +
.../src/main/java/quarks/topology/TStream.java | 32 +++++++++-------
.../src/main/java/quarks/topology/TWindow.java | 5 ++-
.../src/main/java/quarks/topology/Topology.java | 12 ++++--
.../java/quarks/topology/TopologyProvider.java | 3 +-
.../quarks/topology/mbeans/package-info.java | 2 -
.../topology/plumbing/LoadBalancedSplitter.java | 2 +-
.../topology/plumbing/PlumbingStreams.java | 37 ++++++++++++------
.../java/quarks/topology/plumbing/Valve.java | 2 +-
.../java/quarks/topology/tester/Condition.java | 1 +
.../java/quarks/topology/tester/Tester.java | 4 +-
.../java/quarks/test/topology/TStreamTest.java | 9 +++++
.../src/main/java/quarks/window/Partition.java | 3 +-
.../src/main/java/quarks/window/Policies.java | 30 +++++++++++++++
.../src/main/java/quarks/window/Window.java | 10 ++---
.../src/main/java/quarks/window/Windows.java | 6 ++-
.../java/quarks/apps/iot/IotDevicePubSub.java | 4 +-
.../java/quarks/apps/runtime/JobMonitorApp.java | 6 +--
.../quarks/connectors/runtime/Connector.java | 3 +-
.../quarks/connectors/file/FileStreams.java | 6 ++-
.../connectors/file/FileWriterCycleConfig.java | 30 +++++++++++++--
.../connectors/file/FileWriterFlushConfig.java | 26 +++++++++++--
.../connectors/file/FileWriterPolicy.java | 2 +-
.../file/FileWriterRetentionConfig.java | 20 ++++++++--
.../file/runtime/AbstractWriterFile.java | 6 ++-
.../file/runtime/IFileWriterPolicy.java | 6 +--
.../test/connectors/file/FileStreamsTest.java | 1 +
.../quarks/connectors/http/HttpStreams.java | 4 ++
.../connectors/http/runtime/HttpRequester.java | 3 ++
.../quarks/connectors/iot/package-info.java | 4 +-
.../java/quarks/connectors/iotf/IotfDevice.java | 3 +-
.../quarks/connectors/jdbc/JdbcStreams.java | 6 +++
.../quarks/connectors/jdbc/ParameterSetter.java | 2 +-
.../connectors/jdbc/StatementSupplier.java | 2 +
.../quarks/connectors/kafka/KafkaConsumer.java | 8 ++--
.../quarks/connectors/kafka/KafkaProducer.java | 4 +-
.../java/quarks/connectors/mqtt/MqttConfig.java | 28 +++++++-------
.../quarks/connectors/mqtt/MqttStreams.java | 8 +++-
.../connectors/pubsub/PublishSubscribe.java | 4 ++
.../pubsub/service/PublishSubscribeService.java | 2 +
.../websocket/QuarksSslContainerProvider.java | 1 -
.../quarks/connectors/serial/SerialDevice.java | 1 +
.../javax/websocket/WebSocketServerEcho.java | 4 +-
.../java/quarks/console/server/HttpServer.java | 4 +-
.../java/quarks/console/server/ServerUtil.java | 1 +
.../java/quarks/providers/iot/IotProvider.java | 4 +-
.../java/quarks/providers/iot/package-info.java | 2 -
.../quarks/runtime/etiao/AbstractContext.java | 3 ++
.../java/quarks/runtime/etiao/Executable.java | 5 ++-
.../java/quarks/runtime/etiao/Invocation.java | 1 +
.../quarks/runtime/etiao/SettableForwarder.java | 3 +-
.../runtime/etiao/graph/model/GraphType.java | 3 ++
.../runtime/etiao/graph/model/IdMapper.java | 1 +
.../runtime/jmxcontrol/JMXControlService.java | 1 +
.../quarks/runtime/jobregistry/JobEvents.java | 5 ++-
.../runtime/jsoncontrol/JsonControlService.java | 2 +
.../samples/apps/AbstractApplication.java | 3 +-
.../samples/apps/ApplicationUtilities.java | 6 ++-
.../java/quarks/samples/apps/JsonTuples.java | 4 +-
.../apps/sensorAnalytics/package-info.java | 7 ++--
.../java/quarks/samples/connectors/Util.java | 4 +-
.../samples/connectors/iotf/IotfSensors.java | 8 +++-
.../quarks/samples/connectors/jdbc/DbUtils.java | 6 +--
.../samples/connectors/jdbc/PersonData.java | 2 +-
.../samples/connectors/kafka/PublisherApp.java | 1 +
.../quarks/samples/connectors/kafka/Runner.java | 3 +-
.../samples/connectors/kafka/SubscriberApp.java | 1 +
.../samples/connectors/mqtt/PublisherApp.java | 1 +
.../quarks/samples/connectors/mqtt/Runner.java | 2 +-
.../samples/connectors/mqtt/SubscriberApp.java | 1 +
.../samples/console/ConsoleWaterDetector.java | 4 --
.../scenarios/iotf/IotfFullScenario.java | 2 +-
.../CombiningStreamsProcessingResults.java | 2 +
.../quarks/samples/topology/HelloQuarks.java | 2 +
.../quarks/samples/topology/PeriodicSource.java | 2 +
.../samples/topology/SensorsAggregates.java | 2 +
.../utils/sensor/PeriodicRandomSensor.java | 1 +
.../utils/sensor/SimpleSimulatedSensor.java | 21 ++++++----
.../sensor/SimulatedTemperatureSensor.java | 11 ++++--
.../java/quarks/graph/spi/AbstractGraph.java | 2 +
.../quarks/topology/spi/AbstractTopology.java | 1 +
.../topology/spi/graph/GraphTopology.java | 2 +
.../quarks/test/fvt/iot/IotProviderTest.java | 4 +-
.../test/svt/apps/GpsAnalyticsApplication.java | 3 +-
.../src/main/java/quarks/metrics/Metrics.java | 2 +
.../main/java/quarks/metrics/MetricsSetup.java | 5 +++
.../java/quarks/metrics/oplets/CounterOp.java | 2 +
.../java/quarks/metrics/oplets/RateMeter.java | 2 +
.../oplets/SingleMetricAbstractOplet.java | 2 +
.../java/quarks/streamscope/StreamScope.java | 14 +++++--
.../quarks/streamscope/StreamScopeRegistry.java | 28 ++++++++++++--
.../quarks/streamscope/StreamScopeSetup.java | 3 +-
.../streamscope/mbeans/StreamScopeMXBean.java | 4 +-
114 files changed, 519 insertions(+), 174 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/analytics/sensors/src/main/java/quarks/analytics/sensors/Deadtime.java
----------------------------------------------------------------------
diff --git a/analytics/sensors/src/main/java/quarks/analytics/sensors/Deadtime.java b/analytics/sensors/src/main/java/quarks/analytics/sensors/Deadtime.java
index debfe1b..f7abe56 100644
--- a/analytics/sensors/src/main/java/quarks/analytics/sensors/Deadtime.java
+++ b/analytics/sensors/src/main/java/quarks/analytics/sensors/Deadtime.java
@@ -60,7 +60,7 @@ public class Deadtime<T> implements Predicate<T> {
* @param deadtimePeriod the amount of time for {@code test()}
* to return false after returning true.
* Specify a value of 0 for no deadtime period.
- * Must be >= 0.
+ * Must be >= 0.
* @param unit {@link TimeUnit} of {@code deadtimePeriod}
*/
public synchronized void setPeriod(long deadtimePeriod, TimeUnit unit) {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/analytics/sensors/src/main/java/quarks/analytics/sensors/Filters.java
----------------------------------------------------------------------
diff --git a/analytics/sensors/src/main/java/quarks/analytics/sensors/Filters.java b/analytics/sensors/src/main/java/quarks/analytics/sensors/Filters.java
index dedf829..0a71713 100644
--- a/analytics/sensors/src/main/java/quarks/analytics/sensors/Filters.java
+++ b/analytics/sensors/src/main/java/quarks/analytics/sensors/Filters.java
@@ -100,7 +100,6 @@ public class Filters {
* </UL>
* <BR>
* <img src="doc-files/deadband.png" alt="Deadband example">
- * </P>
*
* @param <T> Tuple type.
* @param <V> Value type for the deadband function.
@@ -129,6 +128,7 @@ public class Filters {
* Use {@link Deadtime} directly if you need to change the deadtime period
* while the topology is running.
* </p>
+ * @param <T> tuple type
* @param stream TStream to add deadtime filter to
* @param deadtimePeriod the deadtime period in {@code unit}
* @param unit the {@link TimeUnit} to apply to {@code deadtimePeriod}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/analytics/sensors/src/main/java/quarks/analytics/sensors/Range.java
----------------------------------------------------------------------
diff --git a/analytics/sensors/src/main/java/quarks/analytics/sensors/Range.java b/analytics/sensors/src/main/java/quarks/analytics/sensors/Range.java
index 262940a..e05e70c 100644
--- a/analytics/sensors/src/main/java/quarks/analytics/sensors/Range.java
+++ b/analytics/sensors/src/main/java/quarks/analytics/sensors/Range.java
@@ -163,10 +163,12 @@ public final class Range<T extends Comparable<?>> implements Predicate<T>, Seria
* <p>
* See {@link Ranges} for a collection of convenience constructors.
*
+ * @param <T> a Comparable type
* @param lowerEndpoint null for an infinite value (and lbt must be OPEN)
* @param lbt {@link BoundType} for the lowerEndpoint
* @param upperEndpoint null for an infinite value (and ubt must be OPEN)
* @param ubt {@link BoundType} for the upperEndpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> range(T lowerEndpoint, BoundType lbt, T upperEndpoint, BoundType ubt) {
// matchs Guava Range.range param order
@@ -372,11 +374,13 @@ public final class Range<T extends Comparable<?>> implements Predicate<T>, Seria
* <p>
* See {@link Ranges} for a collection of valueOf methods
* for several types of {@code T}.
- *
+ *
+ * @param <T> a Comparable type
* @param toStringValue value from toString() or has the same syntax.
* @param fromString function to create a T from its String value from
* the parsed toStringValue. Should throw an IllegalArgumentException
* if unable to perform the conversion.
+ * @return the Range
* @throws IllegalArgumentException if unable to parse or convert to
* endpoint in toStringValue to a T.
*/
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/analytics/sensors/src/main/java/quarks/analytics/sensors/Ranges.java
----------------------------------------------------------------------
diff --git a/analytics/sensors/src/main/java/quarks/analytics/sensors/Ranges.java b/analytics/sensors/src/main/java/quarks/analytics/sensors/Ranges.java
index 529b9b3..fddc45a 100644
--- a/analytics/sensors/src/main/java/quarks/analytics/sensors/Ranges.java
+++ b/analytics/sensors/src/main/java/quarks/analytics/sensors/Ranges.java
@@ -32,6 +32,11 @@ public final class Ranges {
* Create a Range (lowerEndpoint..upperEndpoint) (both exclusive/OPEN)
* <p>
* Same as {@code Range.range(BoundType.OPEN, lowerEndpoint, upperEndpoint, BoundType.OPEN)}
+ *
+ * @param <T> Endpoint type
+ * @param lowerEndpoint the endpoint
+ * @param upperEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> open(T lowerEndpoint, T upperEndpoint) {
return Range.range(lowerEndpoint, BoundType.OPEN, upperEndpoint, BoundType.OPEN);
@@ -41,6 +46,11 @@ public final class Ranges {
* Create a Range [lowerEndpoint..upperEndpoint] (both inclusive/CLOSED)
* <p>
* Same as {@code Range.range(BoundType.CLOSED, lowerEndpoint, upperEndpoint, BoundType.CLOSED)}
+ *
+ * @param <T> Endpoint type
+ * @param lowerEndpoint the endpoint
+ * @param upperEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> closed(T lowerEndpoint, T upperEndpoint) {
return Range.range(lowerEndpoint, BoundType.CLOSED, upperEndpoint, BoundType.CLOSED);
@@ -48,6 +58,11 @@ public final class Ranges {
/**
* Create a Range (lowerEndpoint..upperEndpoint] (exclusive/OPEN,inclusive/CLOSED)
+ *
+ * @param <T> Endpoint type
+ * @param lowerEndpoint the endpoint
+ * @param upperEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> openClosed(T lowerEndpoint, T upperEndpoint) {
return Range.range(lowerEndpoint, BoundType.OPEN, upperEndpoint, BoundType.CLOSED);
@@ -55,6 +70,11 @@ public final class Ranges {
/**
* Create a Range [lowerEndpoint..upperEndpoint) (inclusive/CLOSED,exclusive/OPEN)
+ *
+ * @param <T> Endpoint type
+ * @param lowerEndpoint the endpoint
+ * @param upperEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> closedOpen(T lowerEndpoint, T upperEndpoint) {
return Range.range(lowerEndpoint, BoundType.CLOSED, upperEndpoint, BoundType.OPEN);
@@ -62,6 +82,10 @@ public final class Ranges {
/**
* Create a Range (lowerEndpoint..*) (exclusive/OPEN)
+ *
+ * @param <T> Endpoint type
+ * @param lowerEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> greaterThan(T lowerEndpoint) {
return Range.range(lowerEndpoint, BoundType.OPEN, null, BoundType.OPEN);
@@ -69,6 +93,10 @@ public final class Ranges {
/**
* Create a Range [lowerEndpoint..*) (inclusive/CLOSED)
+ *
+ * @param <T> Endpoint type
+ * @param lowerEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> atLeast(T lowerEndpoint) {
return Range.range(lowerEndpoint, BoundType.CLOSED, null, BoundType.OPEN);
@@ -76,6 +104,10 @@ public final class Ranges {
/**
* Create a Range (*..upperEndpoint) (exclusive/OPEN)
+ *
+ * @param <T> Endpoint type
+ * @param upperEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> lessThan(T upperEndpoint) {
return Range.range(null, BoundType.OPEN, upperEndpoint, BoundType.OPEN);
@@ -83,6 +115,10 @@ public final class Ranges {
/**
* Create a Range (*..upperEndpoint] (inclusive/CLOSED)
+ *
+ * @param <T> Endpoint type
+ * @param upperEndpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> atMost(T upperEndpoint) {
return Range.range(null, BoundType.OPEN, upperEndpoint, BoundType.CLOSED);
@@ -90,6 +126,10 @@ public final class Ranges {
/**
* Create a Range [endpoint..endpoint] (both inclusive/CLOSED)
+ *
+ * @param <T> Endpoint type
+ * @param endpoint the endpoint
+ * @return the Range
*/
public static <T extends Comparable<?>> Range<T> singleton(T endpoint) {
return Range.range(endpoint, BoundType.CLOSED, endpoint, BoundType.CLOSED);
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/execution/src/main/java/quarks/execution/Configs.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/Configs.java b/api/execution/src/main/java/quarks/execution/Configs.java
index 7e1e988..4d80d32 100644
--- a/api/execution/src/main/java/quarks/execution/Configs.java
+++ b/api/execution/src/main/java/quarks/execution/Configs.java
@@ -26,7 +26,7 @@ package quarks.execution;
* {@linkplain quarks.execution.Submitter#submit(Object, com.google.gson.JsonObject) submitted}.
* <p>
* The configuration JSON representation is summarized in the following table:
- *
+ * </p>
* <table border=1 cellpadding=3 cellspacing=1>
* <caption>Summary of configuration properties</caption>
* <tr>
@@ -40,7 +40,6 @@ package quarks.execution;
* <td>The name of the job.</td>
* </tr>
* </table>
- * </p>
*/
public interface Configs {
/**
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/execution/src/main/java/quarks/execution/mbeans/PeriodMXBean.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/mbeans/PeriodMXBean.java b/api/execution/src/main/java/quarks/execution/mbeans/PeriodMXBean.java
index cf030a1..847b77b 100644
--- a/api/execution/src/main/java/quarks/execution/mbeans/PeriodMXBean.java
+++ b/api/execution/src/main/java/quarks/execution/mbeans/PeriodMXBean.java
@@ -49,11 +49,14 @@ public interface PeriodMXBean {
/**
* Set the period.
+ * @param period the period
*/
public void setPeriod(long period);
/**
* Set the period and unit
+ * @param period the period
+ * @param unit the {@link TimeUnit} of {@code period}
*/
public void setPeriod(long period, TimeUnit unit);
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/execution/src/main/java/quarks/execution/services/ControlService.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/services/ControlService.java b/api/execution/src/main/java/quarks/execution/services/ControlService.java
index c7888f6..64d6bd7 100644
--- a/api/execution/src/main/java/quarks/execution/services/ControlService.java
+++ b/api/execution/src/main/java/quarks/execution/services/ControlService.java
@@ -39,7 +39,7 @@ package quarks.execution.services;
* </P>
* <P>
* An instance of a control service MBean is defined by its:
- *
+ * </P>
* <UL>
* <LI> A type </LI>
* <LI> A identifier - Unique within the current execution context.</LI>
@@ -48,6 +48,7 @@ package quarks.execution.services;
* <LI> A Java interface - This defines what operations can be executed
* against the control MBean.</LI>
* </UL>
+ * <P>
* A remote system should be able to specify an operation on an
* control server MBean though its alias and type. For example
* an application might be submitted with a fixed name
@@ -72,7 +73,8 @@ public interface ControlService {
/**
* Register a control MBean.
- *
+ *
+ * @param <T> Control MBean type
* @param type Type of the control MBean.
* @param id
* Unique identifier for the control MBean.
@@ -89,12 +91,14 @@ public interface ControlService {
/**
* Unregister a control bean registered by {@link #registerControl(String, String, String, Class, Object)}
+ * @param controlId control's registration identifier returned by {@code registerControl}
*/
void unregister(String controlId);
/**
* Return a control Mbean registered with this service.
*
+ * @param <T> Control MBean type
* @param type Type of the control MBean.
* @param alias Alias for the control MBean.
* @param controlInterface
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/execution/src/main/java/quarks/execution/services/Controls.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/services/Controls.java b/api/execution/src/main/java/quarks/execution/services/Controls.java
index a41b196..7629b8b 100644
--- a/api/execution/src/main/java/quarks/execution/services/Controls.java
+++ b/api/execution/src/main/java/quarks/execution/services/Controls.java
@@ -45,7 +45,7 @@ public class Controls {
* <LI>Any enumeration</LI>
* </UL>
* </UL>
- * @param controlInterface
+ * @param controlInterface class of the control interface
* @return True
*/
public static boolean isControlServiceMBean(Class<?> controlInterface) {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/execution/src/main/java/quarks/execution/services/JobRegistryService.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/services/JobRegistryService.java b/api/execution/src/main/java/quarks/execution/services/JobRegistryService.java
index d379f17..684a427 100644
--- a/api/execution/src/main/java/quarks/execution/services/JobRegistryService.java
+++ b/api/execution/src/main/java/quarks/execution/services/JobRegistryService.java
@@ -31,12 +31,12 @@ import quarks.function.BiConsumer;
* registered jobs, as well as register listeners which are notified on job
* registrations, removals, and updates.
* The following event types are sent to registered listeners:
+ * </p>
* <ul>
* <li>An {@link EventType#ADD} event is sent when a job is added.</li>
* <li>An {@link EventType#REMOVE} event is sent when a job is removed.</li>
* <li>An {@link EventType#UPDATE} event is sent when a job is updated.</li>
* </ul>
- * <p>
* <h3>Event dispatch</h3>
* If a listener invocation throws an Exception, then the exception
* will not prevent the remaining listeners from being invoked. However,
@@ -89,6 +89,7 @@ public interface JobRegistryService {
/**
* Returns a job given its identifier.
*
+ * @param id job identifier
* @return the job or {@code null} if no job is registered with that
* identifier.
*/
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/execution/src/main/java/quarks/execution/services/RuntimeServices.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/services/RuntimeServices.java b/api/execution/src/main/java/quarks/execution/services/RuntimeServices.java
index be8d950..ecadfa7 100644
--- a/api/execution/src/main/java/quarks/execution/services/RuntimeServices.java
+++ b/api/execution/src/main/java/quarks/execution/services/RuntimeServices.java
@@ -29,6 +29,7 @@ public interface RuntimeServices {
* Get a service for this invocation.
* <P>
* These services must be provided by all implementations:
+ * </P>
* <UL>
* <LI>
* {@code java.util.concurrent.ThreadFactory} - Thread factory, runtime code should
@@ -39,9 +40,8 @@ public interface RuntimeServices {
* execute asynchronous and repeating tasks using this scheduler.
* </LI>
* </UL>
- * </P>
- *
*
+ * @param <T> service class type
* @param serviceClass Type of the service required.
* @return Service of type implementing {@code serviceClass} if the
* container this invocation runs in supports that service,
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/execution/src/main/java/quarks/execution/services/ServiceContainer.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/services/ServiceContainer.java b/api/execution/src/main/java/quarks/execution/services/ServiceContainer.java
index 88add9b..92c58a9 100644
--- a/api/execution/src/main/java/quarks/execution/services/ServiceContainer.java
+++ b/api/execution/src/main/java/quarks/execution/services/ServiceContainer.java
@@ -57,6 +57,7 @@ public class ServiceContainer {
* container has previously contained a service for the class,
* the old value is replaced by the specified value.
*
+ * @param <T> service class type
* @param serviceClass the class of service to add.
* @param service service to add to this container.
* @return the previous value associated with {@code serviceClass}, or
@@ -71,6 +72,7 @@ public class ServiceContainer {
/**
* Removes the specified service from this {@code ServiceContainer}.
*
+ * @param <T> service class type
* @param serviceClass the class of service to remove.
* @return the service previously associated with the service class, or
* {@code null} if there was no registered service.
@@ -85,6 +87,7 @@ public class ServiceContainer {
* mapped, or {@code null} if this {@code ServiceContainer} contains no
* service for that key.
*
+ * @param <T> service class type
* @param serviceClass the class whose associated service is to be returned
* @return the service instance mapped to the specified service class, or
* {@code null} if no service is registered for the class.
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/function/src/main/java/quarks/function/BiConsumer.java
----------------------------------------------------------------------
diff --git a/api/function/src/main/java/quarks/function/BiConsumer.java b/api/function/src/main/java/quarks/function/BiConsumer.java
index 5004a07..1482353 100644
--- a/api/function/src/main/java/quarks/function/BiConsumer.java
+++ b/api/function/src/main/java/quarks/function/BiConsumer.java
@@ -22,6 +22,8 @@ import java.io.Serializable;
/**
* Consumer function that takes two arguments.
+ * @param <T> Type of the first function argument
+ * @param <U> Type of the second function argument
*/
public interface BiConsumer<T,U> extends Serializable {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/function/src/main/java/quarks/function/Functions.java
----------------------------------------------------------------------
diff --git a/api/function/src/main/java/quarks/function/Functions.java b/api/function/src/main/java/quarks/function/Functions.java
index 893be1b..acb271d 100644
--- a/api/function/src/main/java/quarks/function/Functions.java
+++ b/api/function/src/main/java/quarks/function/Functions.java
@@ -47,6 +47,7 @@ public class Functions {
/**
* Returns the identity function that returns its single argument.
+ * @param <T> tuple type
* @return Identity function that returns its single argument.
*/
@SuppressWarnings("unchecked")
@@ -56,6 +57,7 @@ public class Functions {
/**
* Returns a constant function that returns zero (0).
+ * @param <T> tuple type
* @return Constant function that returns zero (0).
*/
@SuppressWarnings("unchecked")
@@ -67,7 +69,8 @@ public class Functions {
* Returns a constant function that returns zero (0).
* This is identical to {@link #zero()} but is more
* readable when applied as a key function.
- * @return Constant function that returns zero (0).
+ * @param <T> tuple type
+ * @return Constant function that returns zero (0).
*/
public static <T> Function<T,Integer> unpartitioned() {
return zero();
@@ -97,6 +100,8 @@ public class Functions {
* If {@code function} implements {@code AutoCloseable} then
* the function is assumed to be stateful and a thread-safe
* version is returned.
+ * @param <T> source tuple type
+ * @param <R> result tuple type
* @param function Function to return a thread-safe version of.
* @return A thread-safe function
*/
@@ -134,6 +139,7 @@ public class Functions {
* If {@code function} implements {@code AutoCloseable} then
* the function is assumed to be stateful and a thread-safe
* version is returned.
+ * @param <T> tuple type
* @param function Function to return a thread-safe version of.
* @return A thread-safe function
*/
@@ -171,6 +177,7 @@ public class Functions {
* If {@code function} implements {@code AutoCloseable} then
* the function is assumed to be stateful and a thread-safe
* version is returned.
+ * @param <T> tuple type
* @param function Function to return a thread-safe version of.
* @return A thread-safe function
*/
@@ -209,6 +216,9 @@ public class Functions {
* If {@code function} implements {@code AutoCloseable} then
* the function is assumed to be stateful and a thread-safe
* version is returned.
+ * @param <T> Type of function's first argument
+ * @param <U> Type of function's second argument
+ * @param <R> Type of function's third argument
* @param function Function to return a thread-safe version of.
* @return A thread-safe function
*/
@@ -330,6 +340,7 @@ public class Functions {
* This can be used to delay the execution of the consumer
* until some time in the future using an executor service.
*
+ * @param <T> tuple type
* @param consumer Function to be applied to {@code value}.
* @param value Value to be consumed.
*
@@ -346,7 +357,7 @@ public class Functions {
* @param finalAction Action to be invoked after {@code action.run()} is called.
* @return {@code Runnable} that invokes {@code action.run()} and then {@code finalAction.run()}
*/
- public static <T> Runnable runWithFinal(Runnable action, Runnable finalAction) {
+ public static Runnable runWithFinal(Runnable action, Runnable finalAction) {
return () -> { try {action.run(); } finally {finalAction.run();}};
}
@@ -358,6 +369,7 @@ public class Functions {
/**
* A Consumer that discards all items passed to it.
+ * @param <T> tuple type
* @return A Consumer that discards all items passed to it.
*/
@SuppressWarnings("unchecked")
@@ -377,6 +389,7 @@ public class Functions {
/**
* A Predicate that is always true
+ * @param <T> tuple type
* @return A Predicate that is always true.
*/
@SuppressWarnings("unchecked")
@@ -386,6 +399,7 @@ public class Functions {
/**
* A Predicate that is always false
+ * @param <T> tuple type
* @return A Predicate that is always false.
*/
@SuppressWarnings("unchecked")
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/function/src/main/java/quarks/function/WrappedFunction.java
----------------------------------------------------------------------
diff --git a/api/function/src/main/java/quarks/function/WrappedFunction.java b/api/function/src/main/java/quarks/function/WrappedFunction.java
index 5132fe2..4519052 100644
--- a/api/function/src/main/java/quarks/function/WrappedFunction.java
+++ b/api/function/src/main/java/quarks/function/WrappedFunction.java
@@ -42,6 +42,7 @@ public abstract class WrappedFunction<F> implements Serializable {
/**
* Unwrap to find the outermost function that is an instance of {@code clazz}.
+ * @param <C> Type of class.
* @param clazz Implementation class to search for.
* @return outermost function that is an instance of {@code clazz},
* {@code null} if {@code clazz} is not implemented by {@code this}
@@ -58,6 +59,7 @@ public abstract class WrappedFunction<F> implements Serializable {
* If a function object is not an instance of {@code clazz} but is an instance of
* {@code WrappedFunction} then the test is repeated on the value of {@link #f()}.
*
+ * @param <C> Type of class.
* @param clazz Implementation class to search for.
* @param wf Function to unwrap
* @return outermost function that implements {@code clazz}, {@code null} if
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/graph/src/main/java/quarks/graph/Connector.java
----------------------------------------------------------------------
diff --git a/api/graph/src/main/java/quarks/graph/Connector.java b/api/graph/src/main/java/quarks/graph/Connector.java
index 59002a6..2cf9ca1 100644
--- a/api/graph/src/main/java/quarks/graph/Connector.java
+++ b/api/graph/src/main/java/quarks/graph/Connector.java
@@ -93,7 +93,8 @@ public interface Connector<T> {
* connections. This method may be called multiple times to insert multiple
* peeks. Each tuple submitted to this output port will be seen by all peeks
* in order of their insertion, starting with the first peek inserted.
- *
+ *
+ * @param <N> Peek oplet type
* @param oplet
* Oplet to insert.
* @return {@code output}
@@ -122,11 +123,11 @@ public interface Connector<T> {
* <p>
* The alias must be unique within the topology.
* The alias may be used in various contexts:
+ * </p>
* <ul>
* <li>Runtime control services for the Connector (stream/outputport)
* are registered with this alias.</li>
* </ul>
- * </p>
*
* @param alias the alias
* @throws IllegalStateException if the an alias has already been set
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/graph/src/main/java/quarks/graph/Graph.java
----------------------------------------------------------------------
diff --git a/api/graph/src/main/java/quarks/graph/Graph.java b/api/graph/src/main/java/quarks/graph/Graph.java
index 7a0dc0a..7e01890 100644
--- a/api/graph/src/main/java/quarks/graph/Graph.java
+++ b/api/graph/src/main/java/quarks/graph/Graph.java
@@ -43,6 +43,9 @@ public interface Graph {
* Add a new unconnected {@code Vertex} into the graph.
* <p>
*
+ * @param <N> an Oplet type
+ * @param <C> tuple type of input streams
+ * @param <P> tuple type of output streams
* @param oplet the oplet to associate with the new vertex
* @param inputs the number of input connectors for the new vertex
* @param outputs the number of output connectors for the new vertex
@@ -55,6 +58,8 @@ public interface Graph {
* specified source {@link Oplet}.
* <p>
* The {@code Vertex} for the oplet has 0 input connectors and one output connector.
+ * @param <N> a Source type
+ * @param <P> tuple type
* @param oplet the source oplet
* @return the output connector for the newly created vertex.
*/
@@ -68,7 +73,10 @@ public interface Graph {
* An {@link Edge} is created connecting the specified output connector to
* the new vertice's input connector.
*
- * @param output
+ * @param <N> an Oplet type
+ * @param <C> tuple type of input streams
+ * @param <P> tuple type of output streams
+ * @param output the output connector to connect to the vertice's input connector
* @param oplet the oplet to associate with the new {@code Vertex}
* @return the output connector for the new {@code Vertex}
*/
@@ -96,6 +104,7 @@ public interface Graph {
/**
* Return an unmodifiable view of all edges in this graph.
+ * @return the collection
*/
Collection<Edge> getEdges();
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/oplet/src/main/java/quarks/oplet/Oplet.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/quarks/oplet/Oplet.java b/api/oplet/src/main/java/quarks/oplet/Oplet.java
index fac6d00..5f5af81 100644
--- a/api/oplet/src/main/java/quarks/oplet/Oplet.java
+++ b/api/oplet/src/main/java/quarks/oplet/Oplet.java
@@ -38,8 +38,8 @@ public interface Oplet<I, O> extends AutoCloseable {
/**
* Initialize the oplet.
*
- * @param context
- * @throws Exception
+ * @param context the OpletContext
+ * @throws Exception on failure
*/
void initialize(OpletContext<I, O> context) throws Exception;
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/oplet/src/main/java/quarks/oplet/OpletContext.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/quarks/oplet/OpletContext.java b/api/oplet/src/main/java/quarks/oplet/OpletContext.java
index 3c19651..930539e 100644
--- a/api/oplet/src/main/java/quarks/oplet/OpletContext.java
+++ b/api/oplet/src/main/java/quarks/oplet/OpletContext.java
@@ -31,8 +31,8 @@ import quarks.function.Consumer;
* {@link #getOutputs() output ports} for tuple submission
* and {@link #getJobContext() job} information.
*
- * @param <I>
- * @param <O>
+ * @param <I> tuple type of input streams
+ * @param <O> tuple type of output streams
*/
public interface OpletContext<I, O> extends RuntimeServices {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/oplet/src/main/java/quarks/oplet/core/FanIn.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/quarks/oplet/core/FanIn.java b/api/oplet/src/main/java/quarks/oplet/core/FanIn.java
index b1c5f48..80c1ab0 100644
--- a/api/oplet/src/main/java/quarks/oplet/core/FanIn.java
+++ b/api/oplet/src/main/java/quarks/oplet/core/FanIn.java
@@ -35,6 +35,9 @@ import quarks.oplet.OpletContext;
* {@code receiver} either returns a tuple to emit on the output
* stream or null.
* </P>
+ *
+ * @param <T> Tuple type of input streams
+ * @param <U> Tuple type of output stream
*/
public class FanIn<T,U> extends AbstractOplet<T, U> {
private BiFunction<T, Integer, U> receiver;
@@ -67,7 +70,7 @@ public class FanIn<T,U> extends AbstractOplet<T, U> {
/**
* Set the receiver function. Must be called no later than as part
* of {@link #initialize(OpletContext)}.
- * @param receiver
+ * @param receiver function to receive tuples
*/
protected void setReceiver(BiFunction<T, Integer, U> receiver) {
this.receiver = receiver;
@@ -85,7 +88,7 @@ public class FanIn<T,U> extends AbstractOplet<T, U> {
/**
* Create a Consumer for the input port that invokes the
* receiver and submits a generated tuple, if any, to the output.
- * @param iportIndex
+ * @param iportIndex index of the input port
* @return the Consumer
*/
protected Consumer<T> consumer(int iportIndex) {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/oplet/src/main/java/quarks/oplet/core/Union.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/quarks/oplet/core/Union.java b/api/oplet/src/main/java/quarks/oplet/core/Union.java
index e33257a..9f0c711 100644
--- a/api/oplet/src/main/java/quarks/oplet/core/Union.java
+++ b/api/oplet/src/main/java/quarks/oplet/core/Union.java
@@ -26,6 +26,8 @@ import quarks.oplet.OpletContext;
*
* Processing for each input is identical
* and just submits the tuple to the single output.
+ *
+ * @param <T> Tuple type
*/
public final class Union<T> extends FanIn<T, T> {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/oplet/src/main/java/quarks/oplet/plumbing/PressureReliever.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/quarks/oplet/plumbing/PressureReliever.java b/api/oplet/src/main/java/quarks/oplet/plumbing/PressureReliever.java
index aa6efee..0f240d8 100644
--- a/api/oplet/src/main/java/quarks/oplet/plumbing/PressureReliever.java
+++ b/api/oplet/src/main/java/quarks/oplet/plumbing/PressureReliever.java
@@ -54,6 +54,7 @@ import quarks.window.Windows;
* Tuples are <B>discarded and not</B> submitted to the
* output port if the downstream processing cannot keep up
* the incoming tuple rate.
+ * </P>
* <UL>
* <LI>For a {@link #PressureReliever(int, Function) count}
* {@code PressureReliever} up to last (most recent) {@code N} tuples
@@ -63,9 +64,7 @@ import quarks.window.Windows;
* <BR> If when an input tuple is processed the window partition contains N tuples, then
* the first (oldest) tuple in the partition is discarded before the input tuple is inserted into the window.
* </UL>
- * </P>
* <P>
- * <BR>
* Insertion of the oplet into a stream disconnects the
* upstream processing from the downstream processing,
* so that downstream processing is executed on a different
@@ -133,7 +132,6 @@ public class PressureReliever<T, K> extends Pipe<T, T> {
*
* @param t
* @param v
- * @return
*/
@Override
public void accept(Partition<T, K, LinkedList<T>> partition, T tuple) {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/TSink.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/TSink.java b/api/topology/src/main/java/quarks/topology/TSink.java
index 9a5c92e..db59ca5 100644
--- a/api/topology/src/main/java/quarks/topology/TSink.java
+++ b/api/topology/src/main/java/quarks/topology/TSink.java
@@ -21,6 +21,7 @@ package quarks.topology;
/**
* Termination point (sink) for a stream.
*
+ * @param <T> Tuple type
*/
public interface TSink<T> extends TopologyElement {
/**
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/TStream.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/TStream.java b/api/topology/src/main/java/quarks/topology/TStream.java
index f4d1a44..57888a6 100644
--- a/api/topology/src/main/java/quarks/topology/TStream.java
+++ b/api/topology/src/main/java/quarks/topology/TStream.java
@@ -73,6 +73,7 @@ public interface TStream<T> extends TopologyElement {
* <P>
* Examples of filtering out all empty strings from stream {@code s} of type
* {@code String}
+ * </P>
*
* <pre>
* <code>
@@ -82,8 +83,6 @@ public interface TStream<T> extends TopologyElement {
* </code>
* </pre>
*
- * </P>
- *
* @param predicate
* Filtering logic to be executed against each tuple.
* @return Filtered stream
@@ -101,6 +100,7 @@ public interface TStream<T> extends TopologyElement {
* <P>
* Examples of transforming a stream containing numeric values as
* {@code String} objects into a stream of {@code Double} values.
+ * </P>
*
* <pre>
* <code>
@@ -115,7 +115,7 @@ public interface TStream<T> extends TopologyElement {
* </code>
* </pre>
*
- * </P>
+ * @param <U> Tuple type of output stream
* @param mapper
* Mapping logic to be executed against each tuple.
* @return Stream that will contain tuples of type {@code U} mapped from this
@@ -138,6 +138,7 @@ public interface TStream<T> extends TopologyElement {
* Examples of mapping a stream containing lines of text into a stream
* of words split out from each line. The order of the words in the stream
* will match the order of the words in the lines.
+ * </P>
*
* <pre>
* <code>
@@ -148,7 +149,6 @@ public interface TStream<T> extends TopologyElement {
* </code>
* </pre>
*
- * </P>
* @param <U> Type of mapped input tuples.
* @param mapper
* Mapper logic to be executed against each tuple.
@@ -164,12 +164,12 @@ public interface TStream<T> extends TopologyElement {
* <P>
* For each tuple on the stream, {@code splitter.applyAsInt(tuple)} is
* called. The return value {@code r} determines the destination stream:
+ * </P>
*
* <pre>
* if r < 0 the tuple is discarded
* else it is sent to the stream at position (r % n) in the returned array.
* </pre>
- * </P>
*
* <P>
* Each split {@code TStream} is exposed by the API. The user has full
@@ -183,16 +183,16 @@ public interface TStream<T> extends TopologyElement {
* for its stream. {@code split()} is more efficient. Each tuple is analyzed
* only once by a single {@code splitter} instance to identify the
* destination stream. For example, these are logically equivalent:
- *
+ * </P>
* <pre>
* List<TStream<String>> streams = stream.split(2, tuple -> tuple.length());
*
* TStream<String> stream0 = stream.filter(tuple -> (tuple.length() % 2) == 0);
* TStream<String> stream1 = stream.filter(tuple -> (tuple.length() % 2) == 1);
* </pre>
- * </P>
* <P>
* Example of splitting a stream of log records by their level attribute:
+ * </P>
*
* <pre>
* <code>
@@ -210,7 +210,6 @@ public interface TStream<T> extends TopologyElement {
* splits.get(2). ... // "other" log record processing pipeline
* </code>
* </pre>
- * </P>
*
* @param n
* the number of output streams
@@ -227,6 +226,7 @@ public interface TStream<T> extends TopologyElement {
* Split a stream's tuples among {@code enumClass.size} streams as specified by
* {@code splitter}.
*
+ * @param <E> Enum type
* @param enumClass
* enum data to split
* @param splitter
@@ -257,9 +257,11 @@ public interface TStream<T> extends TopologyElement {
* <p>
* If {@code sinker} implements {@link AutoCloseable}, its {@code close()}
* method will be called when the topology's execution is terminated.
+ * </P>
* <P>
* Example of terminating a stream of {@code String} tuples by printing them
* to {@code System.out}.
+ * </P>
*
* <pre>
* <code>
@@ -268,8 +270,6 @@ public interface TStream<T> extends TopologyElement {
* </code>
* </pre>
*
- * </P>
- *
* @param sinker
* Logic to be executed against each tuple on this stream.
* @return sink element representing termination of this stream.
@@ -328,6 +328,7 @@ public interface TStream<T> extends TopologyElement {
*
* <P>
* Example of modifying a stream {@code String} values by adding the suffix '{@code extra}'.
+ * </P>
*
* <pre>
* <code>
@@ -336,7 +337,6 @@ public interface TStream<T> extends TopologyElement {
* </code>
* </pre>
*
- * </P>
* <P>
* This method is equivalent to
* {@code map(Function<T,T> modifier}).
@@ -432,7 +432,7 @@ public interface TStream<T> extends TopologyElement {
* {@code other}. A stream cannot be unioned with itself, in this case
* {@code this} will be returned.
*
- * @param other
+ * @param other the other stream
* @return A stream that is the union of {@code this} and {@code other}.
*/
TStream<T> union(TStream<T> other);
@@ -473,10 +473,10 @@ public interface TStream<T> extends TopologyElement {
* <p>
* The alias must be unique within the topology.
* The alias may be used in various contexts:
+ * </p>
* <ul>
* <li>Runtime control services for the stream are registered with this alias.</li>
* </ul>
- * </p>
*
* @param alias an alias for the stream.
* @return this
@@ -498,6 +498,9 @@ public interface TStream<T> extends TopologyElement {
* passed into {@code joiner} and the return value is submitted to the
* returned stream. If call returns null then no tuple is submitted.
*
+ * @param <J> Tuple type of result stream
+ * @param <U> Tuple type of window to join with
+ * @param <K> Key type
* @param keyer Key function for this stream to match the window's key.
* @param window Keyed window to join this stream with.
* @param joiner Join function.
@@ -525,6 +528,9 @@ public interface TStream<T> extends TopologyElement {
* passed into {@code joiner} and the return value is submitted to the
* returned stream. If call returns null then no tuple is submitted.
* </P>
+ * @param <J> Tuple type of result stream
+ * @param <U> Tuple type of stream to join with
+ * @param <K> Key type
* @param keyer Key function for this stream
* @param lastStream Stream to join with.
* @param lastStreamKeyer Key function for {@code lastStream}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/TWindow.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/TWindow.java b/api/topology/src/main/java/quarks/topology/TWindow.java
index 27af37c..82b34da 100644
--- a/api/topology/src/main/java/quarks/topology/TWindow.java
+++ b/api/topology/src/main/java/quarks/topology/TWindow.java
@@ -55,7 +55,8 @@ public interface TWindow<T, K> extends TopologyElement {
* Thus the returned stream will contain a sequence of tuples that where the
* most recent tuple represents the most up to date aggregation of a
* partition.
- *
+ *
+ * @param <U> Tuple type
* @param aggregator
* Logic to aggregation a partition.
* @return A stream that contains the latest aggregations of partitions in this window.
@@ -76,11 +77,13 @@ public interface TWindow<T, K> extends TopologyElement {
* the tuples in any subsequent batch. After a partition is batched, its
* contents are cleared.
*
+ * @param <U> Tuple type
* @param batcher
* Logic to aggregation a partition.
* @return A stream that contains the latest aggregations of partitions in this window.
*/
<U> TStream<U> batch(BiFunction<List<T>, K, U> batcher);
+
/**
* Returns the key function used to map tuples to partitions.
* @return Key function used to map tuples to partitions.
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/Topology.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/Topology.java b/api/topology/src/main/java/quarks/topology/Topology.java
index b5115ec..f7d8234 100644
--- a/api/topology/src/main/java/quarks/topology/Topology.java
+++ b/api/topology/src/main/java/quarks/topology/Topology.java
@@ -58,6 +58,7 @@ public interface Topology extends TopologyElement {
* If {@code data} implements {@link AutoCloseable}, its {@code close()}
* method will be called when the topology's execution is terminated.
*
+ * @param <T> Tuple type
* @param data
* Function that produces that data for the stream.
* @return New stream containing the tuples from the iterator returned by
@@ -74,6 +75,7 @@ public interface Topology extends TopologyElement {
* If {@code data} implements {@link AutoCloseable}, its {@code close()}
* method will be called when the topology's execution is terminated.
*
+ * @param <T> Tuple type
* @param data
* Supplier of the tuples.
* @return New stream containing the tuples from calls to {@code data.get()}
@@ -99,6 +101,7 @@ public interface Topology extends TopologyElement {
* The {@code PeriodMXBean} is registered with the {@link ControlService}
* with type {@link TStream#TYPE} and the stream's alias.
* e.g.,
+ * </p>
* <pre>{@code
* Topology t = ...
* TStream<Integer> stream = t.poll(...).alias("myStreamControlAlias");
@@ -109,10 +112,10 @@ public interface Topology extends TopologyElement {
* String alias = pollStream.getAlias();
* PeriodMXBean control = cs.getControl(TStream.TYPE, alias, PeriodMXBean.class);
* control.setPoll(period, unit);
- * }
+ * }
* }</pre>
- * </p>
- *
+ *
+ * @param <T> Tuple type
* @param data
* Function that produces that data for the stream.
* @param period
@@ -144,6 +147,7 @@ public interface Topology extends TopologyElement {
* method will be called when the topology's execution is terminated.
* </P>
*
+ * @param <T> Tuple type
* @param eventSetup handler to receive the {@code eventSubmitter}
* @return New stream containing the tuples added by {@code eventSubmitter.accept(t)}.
*
@@ -162,6 +166,7 @@ public interface Topology extends TopologyElement {
/**
* Declare a stream of objects.
+ * @param <T> Tuple type
* @param values Values that will be present on the stream.
* @return Stream containing all values in {@code values}.
*/
@@ -171,6 +176,7 @@ public interface Topology extends TopologyElement {
/**
* Declare a stream of constants from a collection.
* The returned stream will contain all the tuples in {@code tuples}.
+ * @param <T> Tuple type
* @param tuples Tuples that will be present on the stream.
* @return Stream containing all values in {@code tuples}.
*/
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/TopologyProvider.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/TopologyProvider.java b/api/topology/src/main/java/quarks/topology/TopologyProvider.java
index 3b81dbb..be27aa5 100644
--- a/api/topology/src/main/java/quarks/topology/TopologyProvider.java
+++ b/api/topology/src/main/java/quarks/topology/TopologyProvider.java
@@ -26,7 +26,8 @@ public interface TopologyProvider {
/**
* Create a new topology with a given name.
- *
+ *
+ * @param name name for topology
* @return A new topology.
*/
Topology newTopology(String name);
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/mbeans/package-info.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/mbeans/package-info.java b/api/topology/src/main/java/quarks/topology/mbeans/package-info.java
index 8ee5105..5db7ad1 100644
--- a/api/topology/src/main/java/quarks/topology/mbeans/package-info.java
+++ b/api/topology/src/main/java/quarks/topology/mbeans/package-info.java
@@ -30,7 +30,6 @@ under the License.
*
* <h4>Submit an Application</h4>
* Method: {@link quarks.topology.mbeans.ApplicationServiceMXBean#submit(String, String)}
- * <P>
* <table border=1 cellpadding=3 cellspacing=1>
* <caption>JSON Submit Application</caption>
* <tr>
@@ -77,6 +76,5 @@ under the License.
* job named {@code EngineTemp}.
* <BR>
* {@code {"type":"appService","alias":"quarks","op":"submit","args":["EngineTemp",{}]}}
- * </P>
*/
package quarks.topology.mbeans;
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/plumbing/LoadBalancedSplitter.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/plumbing/LoadBalancedSplitter.java b/api/topology/src/main/java/quarks/topology/plumbing/LoadBalancedSplitter.java
index 746bdb8..6eaf965 100644
--- a/api/topology/src/main/java/quarks/topology/plumbing/LoadBalancedSplitter.java
+++ b/api/topology/src/main/java/quarks/topology/plumbing/LoadBalancedSplitter.java
@@ -30,7 +30,7 @@ import quarks.topology.TStream;
* This is intended to be used as an argument to {@link TStream#split(int, ToIntFunction)}.
* The splitter maintains state for {@code numChannels} splitter channels,
* tracking whether a channel is busy or free.
- * <P></P>
+ * </P><P>
* {@link #applyAsInt(Object) applyAsInt} awaits a free channel, marks
* the channel as busy, and forwards the tuple to the channel. The end
* of the channel's pipeline must call {@link #channelDone(int)} to
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/plumbing/PlumbingStreams.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/plumbing/PlumbingStreams.java b/api/topology/src/main/java/quarks/topology/plumbing/PlumbingStreams.java
index f2e3eb8..01ca30a 100644
--- a/api/topology/src/main/java/quarks/topology/plumbing/PlumbingStreams.java
+++ b/api/topology/src/main/java/quarks/topology/plumbing/PlumbingStreams.java
@@ -62,7 +62,8 @@ public class PlumbingStreams {
* 100ms before forwarding t2. Hence the overall delay between forwarding
* t1 and t2 is 180ms.
* See {@link #blockingThrottle(long, TimeUnit) blockingThrottle}.
- *
+ *
+ * @param <T> Tuple type
* @param stream Stream t
* @param delay Amount of time to delay a tuple.
* @param unit Time unit for {@code delay}.
@@ -216,7 +217,8 @@ public class PlumbingStreams {
* Implementations may throw {@code OutOfMemoryExceptions}
* if the processing against returned stream cannot keep up
* with the arrival rate of tuples on {@code stream}.
- *
+ *
+ * @param <T> Tuple type
* @param stream Stream to be isolated from downstream processing.
* @param ordered {@code true} to maintain arrival order on the returned stream,
* {@code false} to not guaranteed arrival order.
@@ -237,7 +239,8 @@ public class PlumbingStreams {
* </P><P>
* Processing of tuples occurs in the order they were received.
* </P>
- *
+ *
+ * @param <T> Tuple type
* @param stream Stream to be isolated from downstream processing.
* @param queueCapacity size of the queue between {@code stream} and
* the returned stream.
@@ -257,13 +260,13 @@ public class PlumbingStreams {
* from the supplied {@code mappers} and {@code combiner} arguments.
* </P><P>
* That is, it is logically, if not exactly, the same as:
+ * </P>
* <pre>{@code
* List<Function<TStream<T>,TStream<U>>> pipelines = new ArrayList<>();
* for (Function<T,U> mapper : mappers)
* pipelines.add(s -> s.map(mapper));
* concurrent(stream, pipelines, combiner);
* }</pre>
- * </P>
*
* @param <T> Tuple type on input stream.
* @param <U> Tuple type generated by mappers.
@@ -302,6 +305,7 @@ public class PlumbingStreams {
* and then process the next input tuple in the same manner.
* </P><P>
* Logically, instead of doing this:
+ * </P>
* <pre>{@code
* sensorReadings<T> -> A1 -> A2 -> A3 -> results<R>
* }</pre>
@@ -321,7 +325,7 @@ public class PlumbingStreams {
* |-> isolate(1) -> pipeline3 -> |
* . . .
* }</pre>
- * </P><P>
+ * <P>
* The typical use case for this is when an application has a collection
* of independent analytics to perform on each tuple and the analytics
* are sufficiently long running such that performing them concurrently
@@ -332,13 +336,14 @@ public class PlumbingStreams {
* tuples in parallel, each on a replicated processing pipeline.
* </P><P>
* Threadsafety - one of the following must be true:
+ * </P>
* <ul>
* <li>the tuples from {@code stream} are threadsafe</li>
* <li>the {@code pipelines} do not modify the input tuples</li>
* <li>the {@code pipelines} provide their own synchronization controls
* to protect concurrent modifications of the input tuples</li>
* </ul>
- * </P><P>
+ * <P>
* Logically, a thread is allocated for each of the {@code pipelines}.
* The actual degree of concurrency may be {@link TopologyProvider} dependent.
* </P>
@@ -394,6 +399,9 @@ public class PlumbingStreams {
* <P>
* Same as {@code barrier(others, 1)}
* </P>
+ * @param <T> Tuple type
+ * @param streams input streams
+ * @return the output stream
* @see #barrier(List, int)
*/
public static <T> TStream<List<T>> barrier(List<TStream<T>> streams) {
@@ -437,6 +445,8 @@ public class PlumbingStreams {
* <P>
* Same as {@code parallel(stream, width, splitter, (s,ch) -> s.map(t -> mapper.apply(t, ch))}
* </P>
+ * @param <T> Input stream tuple type
+ * @param <U> Result stream tuple type
* @param stream input stream
* @param splitter the tuple channel allocation function
* @param mapper analytic function
@@ -457,7 +467,7 @@ public class PlumbingStreams {
* partitioning tuples among the channels using {@code splitter}.
* Each channel runs a copy of {@code pipeline}.
* The resulting stream is isolated from the upstream parallel channels.
- * <P></P>
+ * </P><P>
* The ordering of tuples in {@code stream} is not maintained in the
* results from {@code parallel}.
* </P><P>
@@ -467,6 +477,7 @@ public class PlumbingStreams {
* A common splitter function is a {@link #roundRobinSplitter(int) roundRobinSplitter}.
* </P><P>
* The generated graph looks like this:
+ * </P>
* <pre>{@code
* -
* |-> isolate(10) -> pipeline-ch1 -> |
@@ -474,7 +485,6 @@ public class PlumbingStreams {
* |-> isolate(10) -> pipeline-ch3 -> |
* . . .
* }</pre>
- * </P>
*
* @param <T> Input stream tuple type
* @param <R> Result stream tuple type
@@ -526,7 +536,7 @@ public class PlumbingStreams {
* partitioning tuples among the channels in a load balanced fashion.
* Each channel runs a copy of {@code pipeline}.
* The resulting stream is isolated from the upstream parallel channels.
- * <P></P>
+ * </P><P>
* The ordering of tuples in {@code stream} is not maintained in the
* results from {@code parallel}.
* </P><P>
@@ -537,6 +547,7 @@ public class PlumbingStreams {
* A {@link LoadBalancedSplitter} is used to distribute tuples.
* </P><P>
* The generated graph looks like this:
+ * </P>
* <pre>{@code
* -
* |-> isolate(1) -> pipeline-ch1 -> peek(splitter.channelDone()) -> |
@@ -544,7 +555,7 @@ public class PlumbingStreams {
* |-> isolate(1) -> pipeline-ch3 -> peek(splitter.channelDone()) -> |
* . . .
* }</pre>
- * </P><P>
+ * <P>
* Note, this implementation requires that the splitter is used from
* only a single JVM. The {@link quarks.providers.direct.DirectProvider DirectProvider}
* provider meets this requirement.
@@ -603,6 +614,9 @@ public class PlumbingStreams {
* on successive calls to {@code roundRobinSplitter.applyAsInt()},
* returning {@code 0, 1, ..., width-1, 0, 1, ..., width-1}.
* </P>
+ * @param <T> Tuple type
+ * @param width number of splitter channels
+ * @return the splitter
* @see TStream#split(int, ToIntFunction) TStream.split
* @see PlumbingStreams#parallel(TStream, int, ToIntFunction, BiFunction) parallel
*/
@@ -633,6 +647,7 @@ public class PlumbingStreams {
* I.e., You want all of the pipelines to start processing a tuple
* at the same time and not start a new tuple until the current
* tuple had been fully processed by each of them:
+ * </P>
* <pre>{@code
* TStream<Integer> readings = ...;
* Semaphore gateControl = new Semaphore(1); // allow the first to pass through
@@ -647,7 +662,7 @@ public class PlumbingStreams {
* );
* TStream<List<Integer>> results = PlumbingStreams.concurrent(gated, pipelines, combiner);
* }</pre>
- * </P>
+ * @param <T> Tuple type
* @param stream the input stream
* @param semaphore gate control
* @return gated stream
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/plumbing/Valve.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/plumbing/Valve.java b/api/topology/src/main/java/quarks/topology/plumbing/Valve.java
index 1200858..51d961e 100644
--- a/api/topology/src/main/java/quarks/topology/plumbing/Valve.java
+++ b/api/topology/src/main/java/quarks/topology/plumbing/Valve.java
@@ -34,6 +34,7 @@ import quarks.function.Predicate;
* <br>
* E.g., in a simple case, a Valve might be used to control
* whether or not logging or publishing of tuples is enabled.
+ * </p>
* <pre>{@code
* TStream<JsonObject> stream = ...;
*
@@ -43,7 +44,6 @@ import quarks.function.Predicate;
* // from some analytic or device command handler...
* valve.setOpen(true);
* }</pre>
- * </p>
*
* @param <T> tuple type
*/
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/tester/Condition.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/tester/Condition.java b/api/topology/src/main/java/quarks/topology/tester/Condition.java
index 00c518d..5d5e2c4 100644
--- a/api/topology/src/main/java/quarks/topology/tester/Condition.java
+++ b/api/topology/src/main/java/quarks/topology/tester/Condition.java
@@ -21,6 +21,7 @@ package quarks.topology.tester;
/**
* Function representing if a condition is valid or not.
*
+ * @param <T> Condition's result type
*/
public interface Condition<T> {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/main/java/quarks/topology/tester/Tester.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/tester/Tester.java b/api/topology/src/main/java/quarks/topology/tester/Tester.java
index df17a4d..7b4c9b2 100644
--- a/api/topology/src/main/java/quarks/topology/tester/Tester.java
+++ b/api/topology/src/main/java/quarks/topology/tester/Tester.java
@@ -99,6 +99,7 @@ public interface Tester extends TopologyElement {
* {@link Condition#valid()} and {@link Condition#getResult()} may change as
* more tuples are seen on {@code stream}. <BR>
*
+ * @param <T> Tuple type
* @param stream
* Stream to be tested.
* @param values
@@ -117,7 +118,8 @@ public interface Tester extends TopologyElement {
* If the topology is still executing then the returned values from
* {@link Condition#valid()} and {@link Condition#getResult()} may change as
* more tuples are seen on {@code stream}. <BR>
- *
+ *
+ * @param <T> Tuple type
* @param stream
* Stream to be tested.
* @param values
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
----------------------------------------------------------------------
diff --git a/api/topology/src/test/java/quarks/test/topology/TStreamTest.java b/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
index b22b2c7..d56f258 100644
--- a/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
+++ b/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
@@ -286,6 +286,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Test split() with no drops.
+ * @throws Exception on failure
*/
@Test
public void testSplit() throws Exception {
@@ -311,6 +312,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Test split() with drops.
+ * @throws Exception on failure
*/
@Test
public void testSplitWithDrops() throws Exception {
@@ -352,6 +354,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Test split() zero outputs
+ * @throws Exception on failure
*/
@Test(expected = IllegalArgumentException.class)
public void testSplitWithZeroOutputs() throws Exception {
@@ -360,6 +363,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Test split() negative outputs
+ * @throws Exception on failure
*/
@Test(expected = IllegalArgumentException.class)
public void testSplitWithNegativeOutputs() throws Exception {
@@ -382,6 +386,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Test split(enum) with integer type enum.
+ * @throws Exception on failure
*/
@Test
public void testSplitWithEnum() throws Exception {
@@ -424,6 +429,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Test split(enum) with integer type enum.
+ * @throws Exception on failure
*/
@Test(expected = IllegalArgumentException.class)
public void testSplitWithEnumForZeroSizeClass() throws Exception {
@@ -704,6 +710,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Submit multiple jobs concurrently using ProcessSource.
+ * @throws Exception on failure
*/
@Test
public void testMultiTopology() throws Exception {
@@ -726,6 +733,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Submit multiple jobs concurrently using ProcessSource.
+ * @throws Exception on failure
*/
@Test
public void testMultiTopologyWithError() throws Exception {
@@ -750,6 +758,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
/**
* Submit multiple jobs concurrently using PeriodicSource.
+ * @throws Exception on failure
*/
@Test
public void testMultiTopologyPollWithError() throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/window/src/main/java/quarks/window/Partition.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Partition.java b/api/window/src/main/java/quarks/window/Partition.java
index 4b80c46..c6ad2c3 100644
--- a/api/window/src/main/java/quarks/window/Partition.java
+++ b/api/window/src/main/java/quarks/window/Partition.java
@@ -26,12 +26,13 @@ import java.util.List;
* returned by {@code getContents} is stable when synchronizing
* on the partition object. For example:
*
- * <pre><code>
+ * <pre>{@code
* Partition<Integer, Integer, ArrayList<Integer>> part = ...;
* synchronized(part){
* List<Integer> = part.getContents();
* // stable operation on contents of partition
* }
+ * }</pre>
*
* @param <T> Type of tuples in the partition.
* @param <K> Type of the partition's key.
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/window/src/main/java/quarks/window/Policies.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Policies.java b/api/window/src/main/java/quarks/window/Policies.java
index 6c2241a..2d3cbbc 100644
--- a/api/window/src/main/java/quarks/window/Policies.java
+++ b/api/window/src/main/java/quarks/window/Policies.java
@@ -37,6 +37,9 @@ public class Policies {
* A policy which schedules a future partition eviction if the partition is empty.
* This can be used as a contents policy that is scheduling the eviction of
* the tuple just about to be inserted.
+ * @param <T> Tuple Type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
* @param time The time span in which tuple are permitted in the partition.
* @param unit The units of time.
* @return The time-based contents policy.
@@ -54,6 +57,9 @@ public class Policies {
* A policy which schedules a future partition eviction on the first insert.
* This can be used as a contents policy that schedules the eviction of tuples
* as a batch.
+ * @param <T> Tuple Type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
* @param time The time span in which tuple are permitted in the partition.
* @param unit The units of time.
* @return The time-based contents policy.
@@ -79,6 +85,8 @@ public class Policies {
* An eviction policy which evicts all tuples that are older than a specified time.
* If any tuples remain in the partition, it schedules their eviction after
* an appropriate interval.
+ * @param <T> Tuple Type
+ * @param <K> Key type
* @param time The timespan in which tuple are permitted in the partition.
* @param unit The units of time.
* @return The time-based eviction policy.
@@ -105,6 +113,8 @@ public class Policies {
/**
* An eviction policy which processes the window, evicts all tuples, and
* schedules the next eviction after the appropriate interval.
+ * @param <T> Tuple Type
+ * @param <K> Key type
* @param time The timespan in which tuple are permitted in the partition.
* @param unit The units of time.
* @return The time-based eviction policy.
@@ -143,6 +153,10 @@ public class Policies {
* If, when called, the number of tuples in the partition is
* greater than equal to {@code count} then {@code partition.evict()}
* is called.
+ * @param <T> Tuple type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
+ * @param count the count
* @return A count-based contents policy.
*/
public static <T, K, L extends List<T>> BiConsumer<Partition<T, K, L>, T> countContentsPolicy(final int count){
@@ -155,6 +169,9 @@ public class Policies {
/**
* Returns a Consumer representing an evict determiner that evict all tuples
* from the window.
+ * @param <T> Tuple type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
* @return An evict determiner that evicts all tuples.
*/
public static <T, K, L extends List<T>> Consumer<Partition<T, K, L> > evictAll(){
@@ -163,6 +180,9 @@ public class Policies {
/**
* Returns an evict determiner that evicts the oldest tuple.
+ * @param <T> Tuple type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
* @return A evict determiner that evicts the oldest tuple.
*/
public static <T, K, L extends List<T>> Consumer<Partition<T, K, L> > evictOldest(){
@@ -172,6 +192,9 @@ public class Policies {
/**
* Returns a trigger policy that triggers
* processing on every insert.
+ * @param <T> Tuple type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
* @return A trigger policy that triggers processing on every insert.
*/
public static <T, K, L extends List<T>> BiConsumer<Partition<T, K, L>, T> processOnInsert(){
@@ -181,6 +204,10 @@ public class Policies {
/**
* Returns a trigger policy that triggers when the size of a partition
* equals or exceeds a value, and then evicts its contents.
+ * @param <T> Tuple type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
+ * @param size partition size
* @return A trigger policy that triggers processing when the size of
* the partition equals or exceets a value.
*/
@@ -195,6 +222,9 @@ public class Policies {
/**
* A {@link BiConsumer} policy which does nothing.
+ * @param <T> Tuple type
+ * @param <K> Key type
+ * @param <L> List type for the partition contents.
* @return A policy which does nothing.
*/
public static <T, K, L extends List<T>> BiConsumer<Partition<T, K, L>, T> doNothing(){
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/window/src/main/java/quarks/window/Window.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Window.java b/api/window/src/main/java/quarks/window/Window.java
index a9ac191..8bf6706 100644
--- a/api/window/src/main/java/quarks/window/Window.java
+++ b/api/window/src/main/java/quarks/window/Window.java
@@ -77,20 +77,20 @@ public interface Window<T, K, L extends List<T>>{
*
*
*
- * @param tuple
+ * @param tuple the tuple to insert
* @return true, if the tuple was successfully inserted. Otherwise, false.
*/
boolean insert(T tuple);
/**
* Register a WindowProcessor.
- * @param windowProcessor
+ * @param windowProcessor function to process the window
*/
void registerPartitionProcessor(BiConsumer<List<T>, K> windowProcessor);
/**
* Register a ScheduledExecutorService.
- * @param ses
+ * @param ses the service
*/
void registerScheduledExecutorService(ScheduledExecutorService ses);
@@ -163,12 +163,12 @@ public interface Window<T, K, L extends List<T>>{
* is stable when synchronizing on the intrinsic lock of the map,
* for example:
* <br>
- * <pre><code>
+ * <pre>{@code
* Map<K, Partitions<U, K, ?>> partitions = window.getPartitions();
* synchronized(partitions){
* // operations with partition
* }
- * </code></pre>
+ * }</pre>
*
* @return A map of the window's keys and partitions.
*/
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/api/window/src/main/java/quarks/window/Windows.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Windows.java b/api/window/src/main/java/quarks/window/Windows.java
index d55af0e..8313e40 100644
--- a/api/window/src/main/java/quarks/window/Windows.java
+++ b/api/window/src/main/java/quarks/window/Windows.java
@@ -40,7 +40,11 @@ public class Windows {
/**
* Create a window using the passed in policies.
- *
+ *
+ * @param <T> Tuple type.
+ * @param <K> Key type.
+ * @param <L> List type.
+ *
* @param insertionPolicy Policy indicating if a tuple should be inserted
* into the window.
* @param contentsPolicy Contents policy called prior to insertion of a tuple.
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/apps/iot/src/main/java/quarks/apps/iot/IotDevicePubSub.java
----------------------------------------------------------------------
diff --git a/apps/iot/src/main/java/quarks/apps/iot/IotDevicePubSub.java b/apps/iot/src/main/java/quarks/apps/iot/IotDevicePubSub.java
index 14b282d..c8d1341 100644
--- a/apps/iot/src/main/java/quarks/apps/iot/IotDevicePubSub.java
+++ b/apps/iot/src/main/java/quarks/apps/iot/IotDevicePubSub.java
@@ -55,12 +55,12 @@ import quarks.topology.TopologyElement;
* <P>
* At startup this application sends a single device event with
* identifier {@link Events#IOT_START}. This performs two functions:
+ * </P>
* <UL>
* <LI>Initiates a connection to the message hub.</LI>
* <LI>Allows external applications to be notified (by subscribing to device events)
* when a Quarks provider starts.</LI>
* </UL>
- * </P>
*
* @see PublishSubscribe
*/
@@ -92,6 +92,8 @@ public class IotDevicePubSub {
/**
* Create an instance of this application using {@code device} as the device
* connection to a message hub.
+ *
+ * @param device the IotDevice
*/
public static void createApplication(IotDevice device) {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/apps/runtime/src/main/java/quarks/apps/runtime/JobMonitorApp.java
----------------------------------------------------------------------
diff --git a/apps/runtime/src/main/java/quarks/apps/runtime/JobMonitorApp.java b/apps/runtime/src/main/java/quarks/apps/runtime/JobMonitorApp.java
index 7e9dfaf..03d060b 100644
--- a/apps/runtime/src/main/java/quarks/apps/runtime/JobMonitorApp.java
+++ b/apps/runtime/src/main/java/quarks/apps/runtime/JobMonitorApp.java
@@ -54,6 +54,7 @@ import quarks.topology.services.ApplicationService;
* <P>
* The monitoring application must be submitted within a context which
* provides the following services:
+ * </P>
* <ul>
* <li>ApplicationService - an {@code ApplicationServiceMXBean} control
* registered by this service is used to resubmit failed applications.</li>
@@ -62,7 +63,6 @@ import quarks.topology.services.ApplicationService;
* restarting failed applications.</li>
* <li>JobRegistryService - generates job monitoring events. </li>
* </ul>
- * </P>
*/
public class JobMonitorApp {
/**
@@ -101,8 +101,8 @@ public class JobMonitorApp {
* Submits the application topology.
*
* @return the job.
- * @throws InterruptedException
- * @throws ExecutionException
+ * @throws InterruptedException if the operation was interrupted
+ * @throws ExecutionException on task execution exception
*/
public Job submit() throws InterruptedException, ExecutionException {
Future<Job> f = submitter.submit(topology);
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/connectors/common/src/main/java/quarks/connectors/runtime/Connector.java
----------------------------------------------------------------------
diff --git a/connectors/common/src/main/java/quarks/connectors/runtime/Connector.java b/connectors/common/src/main/java/quarks/connectors/runtime/Connector.java
index e122d07..8709673 100644
--- a/connectors/common/src/main/java/quarks/connectors/runtime/Connector.java
+++ b/connectors/common/src/main/java/quarks/connectors/runtime/Connector.java
@@ -440,7 +440,8 @@ public abstract class Connector<T> implements AutoCloseable, Serializable {
protected abstract void doClose(T client) throws Exception;
/**
- * Get a connector id to use in log and exception msgs
+ * Get a connector id to use in log and exception msgs
+ * @return the id
*/
protected abstract String id();
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/connectors/file/src/main/java/quarks/connectors/file/FileStreams.java
----------------------------------------------------------------------
diff --git a/connectors/file/src/main/java/quarks/connectors/file/FileStreams.java b/connectors/file/src/main/java/quarks/connectors/file/FileStreams.java
index 119204a..cd80684 100644
--- a/connectors/file/src/main/java/quarks/connectors/file/FileStreams.java
+++ b/connectors/file/src/main/java/quarks/connectors/file/FileStreams.java
@@ -54,6 +54,7 @@ public class FileStreams {
* <p>
* This is the same as {@code directoryWatcher(t, () -> dir, null)}.
*
+ * @param te topology element whose topology the watcher will be added to
* @param directory
* Name of the directory to watch.
* @return Stream containing absolute pathnames of newly created files in
@@ -94,7 +95,8 @@ public class FileStreams {
* MacOs Java lacks a native implementation of {@link WatchService}.
* The result can be a delay in detecting newly created files (e.g., 10sec)
* as well not detecting rapid deletion and recreation of a file.
- *
+ *
+ * @param te topology element whose topology the watcher will be added to
* @param directory
* Name of the directory to watch.
* @param comparator
@@ -209,6 +211,7 @@ public class FileStreams {
* }</pre>
* @param contents the lines to write
* @param basePathname the base pathname of the created files
+ * @return a TSink
*/
public static TSink<String> textFileWriter(TStream<String> contents,
Supplier<String> basePathname) {
@@ -239,6 +242,7 @@ public class FileStreams {
* @param contents the lines to write
* @param basePathname the base pathname of the created files
* @param policy the policy to use. may be null.
+ * @return a TSink
* @see FileWriterPolicy
*/
public static TSink<String> textFileWriter(TStream<String> contents,
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/d7840011/connectors/file/src/main/java/quarks/connectors/file/FileWriterCycleConfig.java
----------------------------------------------------------------------
diff --git a/connectors/file/src/main/java/quarks/connectors/file/FileWriterCycleConfig.java b/connectors/file/src/main/java/quarks/connectors/file/FileWriterCycleConfig.java
index 0c927f9..6958f84 100644
--- a/connectors/file/src/main/java/quarks/connectors/file/FileWriterCycleConfig.java
+++ b/connectors/file/src/main/java/quarks/connectors/file/FileWriterCycleConfig.java
@@ -43,25 +43,45 @@ public class FileWriterCycleConfig<T> {
private long periodMsec;
private Predicate<T> tuplePredicate;
- /** same as {@code newConfig(fileSize, 0, 0, null)} */
+ /** same as {@code newConfig(fileSize, 0, 0, null)}
+ *
+ * @param <T> Tuple type
+ * @param fileSize cycle after {@code fileSize} bytes have been written. 0 to disable.
+ * @return the cycle configuration
+ */
public static <T> FileWriterCycleConfig<T> newFileSizeBasedConfig(long fileSize) {
if (fileSize < 1)
throw new IllegalArgumentException("fileSize");
return newConfig(fileSize, 0, 0, null);
}
- /** same as {@code newConfig0, cntTuples, 0, null)} */
+ /** same as {@code newConfig(0, cntTuples, 0, null)}
+ *
+ * @param <T> Tuple type
+ * @param cntTuples cycle after every {@code cntTuple} tuples have been written. 0 to disable.
+ * @return the cycle configuration
+ */
public static <T> FileWriterCycleConfig<T> newCountBasedConfig(int cntTuples) {
if (cntTuples < 1)
throw new IllegalArgumentException("cntTuples");
return newConfig(0, cntTuples, 0, null);
}
- /** same as {@code newConfig(0, 0, periodMsec, null)} */
+ /** same as {@code newConfig(0, 0, periodMsec, null)}
+ *
+ * @param <T> Tuple type
+ * @param periodMsec cycle after {@code periodMsec} has elapsed since the last time based cycle. 0 to disable.
+ * @return the cycle configuration
+ */
public static <T> FileWriterCycleConfig<T> newTimeBasedConfig(long periodMsec) {
if (periodMsec < 1)
throw new IllegalArgumentException("periodMsec");
return newConfig(0, 0, periodMsec, null);
}
- /** same as {@code newConfig(0, 0, 0, tuplePredicate)} */
+ /** same as {@code newConfig(0, 0, 0, tuplePredicate)}
+ *
+ * @param <T> Tuple type
+ * @param tuplePredicate cycle if {@code tuplePredicate} returns true. null to disable.
+ * @return the cycle configuration
+ */
public static <T> FileWriterCycleConfig<T> newPredicateBasedConfig(Predicate<T> tuplePredicate) {
return newConfig(0, 0, 0, tuplePredicate);
}
@@ -70,10 +90,12 @@ public class FileWriterCycleConfig<T> {
* Create a new configuration.
* <p>
* At least one configuration mode must be enabled.
+ * @param <T> Tuple type
* @param fileSize cycle after {@code fileSize} bytes have been written. 0 to disable.
* @param cntTuples cycle after every {@code cntTuple} tuples have been written. 0 to disable.
* @param periodMsec cycle after {@code periodMsec} has elapsed since the last time based cycle. 0 to disable.
* @param tuplePredicate cycle if {@code tuplePredicate} returns true. null to disable.
+ * @return the cycle configuration
*/
public static <T> FileWriterCycleConfig<T> newConfig(long fileSize, int cntTuples, long periodMsec, Predicate<T> tuplePredicate) {
return new FileWriterCycleConfig<>(fileSize, cntTuples, periodMsec, tuplePredicate);