NIFI-271 checkpoint push because there are so many changes. Long way to go but got through dto library
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/9faaef8c Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/9faaef8c Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/9faaef8c Branch: refs/heads/NIFI-271 Commit: 9faaef8cfae703c5b3a77796620c03787767cc28 Parents: 8a296aa Author: joewitt <[email protected]> Authored: Tue Apr 21 02:15:46 2015 -0400 Committer: joewitt <[email protected]> Committed: Tue Apr 21 02:15:46 2015 -0400 ---------------------------------------------------------------------- nifi-parent/pom.xml | 19 +- .../apache/nifi/bootstrap/BootstrapCodec.java | 118 +- .../org/apache/nifi/bootstrap/NiFiListener.java | 196 +-- .../java/org/apache/nifi/bootstrap/RunNiFi.java | 1641 +++++++++--------- .../org/apache/nifi/bootstrap/ShutdownHook.java | 120 +- .../exception/InvalidCommandException.java | 33 +- .../apache/nifi/util/FlowFilePackagerV1.java | 2 +- .../apache/nifi/util/FlowFileUnpackagerV1.java | 2 +- .../nifi-hl7-query-language/pom.xml | 204 +-- .../src/checkstyle/suppressions.xml | 22 + .../org/apache/nifi/hl7/hapi/EmptyField.java | 16 +- .../org/apache/nifi/hl7/hapi/HapiField.java | 95 +- .../org/apache/nifi/hl7/hapi/HapiMessage.java | 115 +- .../org/apache/nifi/hl7/hapi/HapiSegment.java | 70 +- .../apache/nifi/hl7/hapi/SingleValueField.java | 31 +- .../java/org/apache/nifi/hl7/io/HL7Reader.java | 4 +- .../hl7/io/exception/InvalidHL7Exception.java | 29 +- .../org/apache/nifi/hl7/model/HL7Component.java | 6 +- .../org/apache/nifi/hl7/model/HL7Field.java | 1 - .../org/apache/nifi/hl7/model/HL7Message.java | 8 +- .../org/apache/nifi/hl7/model/HL7Segment.java | 8 +- .../org/apache/nifi/hl7/query/Declaration.java | 12 +- .../org/apache/nifi/hl7/query/HL7Query.java | 646 ++++--- .../org/apache/nifi/hl7/query/QueryResult.java | 15 +- .../org/apache/nifi/hl7/query/ResultHit.java | 7 +- .../org/apache/nifi/hl7/query/Selection.java | 31 +- .../hl7/query/evaluator/BooleanEvaluator.java | 6 +- .../nifi/hl7/query/evaluator/Evaluator.java | 11 +- .../hl7/query/evaluator/IntegerEvaluator.java | 7 +- .../hl7/query/evaluator/StringEvaluator.java | 8 +- .../comparison/AbstractComparisonEvaluator.java | 163 +- .../comparison/AbstractNumericComparison.java | 85 +- .../evaluator/comparison/EqualsEvaluator.java | 16 +- .../comparison/GreaterThanEvaluator.java | 16 +- .../comparison/GreaterThanOrEqualEvaluator.java | 16 +- .../evaluator/comparison/IsNullEvaluator.java | 83 +- .../evaluator/comparison/LessThanEvaluator.java | 15 +- .../comparison/LessThanOrEqualEvaluator.java | 15 +- .../comparison/NotEqualsEvaluator.java | 16 +- .../evaluator/comparison/NotEvaluator.java | 23 +- .../evaluator/comparison/NotNullEvaluator.java | 75 +- .../literal/IntegerLiteralEvaluator.java | 22 +- .../literal/StringLiteralEvaluator.java | 21 +- .../hl7/query/evaluator/logic/AndEvaluator.java | 37 +- .../hl7/query/evaluator/logic/OrEvaluator.java | 37 +- .../message/DeclaredReferenceEvaluator.java | 33 +- .../query/evaluator/message/DotEvaluator.java | 111 +- .../query/evaluator/message/FieldEvaluator.java | 69 +- .../evaluator/message/MessageEvaluator.java | 12 +- .../evaluator/message/SegmentEvaluator.java | 41 +- .../exception/HL7QueryParsingException.java | 33 +- .../nifi/hl7/query/result/MissedResult.java | 59 +- .../hl7/query/result/StandardQueryResult.java | 75 +- .../hl7/query/result/StandardResultHit.java | 31 +- .../org/apache/nifi/hl7/query/TestHL7Query.java | 524 +++--- .../org/apache/nifi/util/NiFiProperties.java | 14 +- .../nifi/security/util/CertificateUtils.java | 21 +- .../nifi/security/util/SecurityStoreTypes.java | 8 +- .../nifi/security/util/SslContextFactory.java | 36 +- .../flowfile/attributes/CoreAttributes.java | 27 +- .../attributes/FlowFileAttributeKey.java | 1 + .../apache/nifi/remote/VersionNegotiator.java | 10 +- .../nifi/remote/io/CompressionInputStream.java | 5 +- .../nifi/remote/io/CompressionOutputStream.java | 2 +- .../remote/io/socket/BufferStateManager.java | 2 +- .../io/socket/SocketChannelInputStream.java | 9 +- .../io/socket/SocketChannelOutputStream.java | 3 +- .../remote/io/socket/ssl/SSLSocketChannel.java | 4 +- .../socket/ssl/SSLSocketChannelInputStream.java | 2 +- .../nifi/stream/io/ByteArrayOutputStream.java | 6 +- .../stream/io/ByteCountingOutputStream.java | 2 - .../stream/io/LeakyBucketStreamThrottler.java | 8 +- .../nifi/stream/io/LimitingInputStream.java | 2 +- .../stream/io/MinimumLengthInputStream.java | 129 +- .../org/apache/nifi/stream/io/StreamUtils.java | 52 +- .../io/util/NonThreadSafeCircularBuffer.java | 2 +- .../java/org/apache/nifi/util/EscapeUtils.java | 15 +- .../java/org/apache/nifi/util/FormatUtils.java | 16 +- .../apache/nifi/util/NaiveSearchRingBuffer.java | 14 +- .../java/org/apache/nifi/util/ObjectHolder.java | 1 - .../java/org/apache/nifi/util/RingBuffer.java | 19 +- .../java/org/apache/nifi/util/StopWatch.java | 8 +- .../main/java/org/apache/nifi/util/Tuple.java | 6 - .../concurrency/DebugDisabledTimedLock.java | 11 +- .../util/concurrency/DebugEnabledTimedLock.java | 16 +- .../org/apache/nifi/util/file/FileUtils.java | 57 +- .../file/monitor/SynchronousFileWatcher.java | 4 +- .../org/apache/nifi/util/search/Search.java | 6 +- .../org/apache/nifi/util/search/SearchTerm.java | 14 +- .../file/monitor/TestCompoundUpdateMonitor.java | 1 - .../monitor/TestSynchronousFileWatcher.java | 1 - .../nifi/util/timebuffer/TestRingBuffer.java | 6 +- .../org/apache/nifi/web/util/ClientUtils.java | 38 +- nifi/nifi-docs/pom.xml | 9 + .../nifi/web/api/dto/BulletinBoardDTO.java | 8 +- .../apache/nifi/web/api/dto/BulletinDTO.java | 38 +- .../nifi/web/api/dto/BulletinQueryDTO.java | 24 +- .../org/apache/nifi/web/api/dto/ClusterDTO.java | 8 +- .../nifi/web/api/dto/ComponentHistoryDTO.java | 8 +- .../apache/nifi/web/api/dto/ConnectableDTO.java | 36 +- .../apache/nifi/web/api/dto/ConnectionDTO.java | 24 +- .../web/api/dto/ControllerConfigurationDTO.java | 44 +- .../apache/nifi/web/api/dto/ControllerDTO.java | 56 +- .../nifi/web/api/dto/ControllerServiceDTO.java | 68 +- ...ontrollerServiceReferencingComponentDTO.java | 69 +- .../org/apache/nifi/web/api/dto/CounterDTO.java | 16 +- .../apache/nifi/web/api/dto/CountersDTO.java | 8 +- .../nifi/web/api/dto/DocumentedTypeDTO.java | 12 +- .../apache/nifi/web/api/dto/FlowSnippetDTO.java | 37 +- .../org/apache/nifi/web/api/dto/LabelDTO.java | 12 +- .../nifi/web/api/dto/NiFiComponentDTO.java | 5 +- .../org/apache/nifi/web/api/dto/NodeDTO.java | 40 +- .../apache/nifi/web/api/dto/NodeEventDTO.java | 12 +- .../web/api/dto/NodeSystemDiagnosticsDTO.java | 8 +- .../org/apache/nifi/web/api/dto/PortDTO.java | 34 +- .../apache/nifi/web/api/dto/PositionDTO.java | 8 +- .../nifi/web/api/dto/PreviousValueDTO.java | 12 +- .../nifi/web/api/dto/ProcessGroupDTO.java | 46 +- .../nifi/web/api/dto/ProcessorConfigDTO.java | 59 +- .../apache/nifi/web/api/dto/ProcessorDTO.java | 22 +- .../nifi/web/api/dto/PropertyDescriptorDTO.java | 63 +- .../nifi/web/api/dto/PropertyHistoryDTO.java | 4 +- .../nifi/web/api/dto/RelationshipDTO.java | 12 +- .../api/dto/RemoteProcessGroupContentsDTO.java | 8 +- .../nifi/web/api/dto/RemoteProcessGroupDTO.java | 70 +- .../web/api/dto/RemoteProcessGroupPortDTO.java | 42 +- .../nifi/web/api/dto/ReportingTaskDTO.java | 272 ++- .../apache/nifi/web/api/dto/RevisionDTO.java | 5 +- .../org/apache/nifi/web/api/dto/SnippetDTO.java | 68 +- .../nifi/web/api/dto/SystemDiagnosticsDTO.java | 116 +- .../apache/nifi/web/api/dto/TemplateDTO.java | 24 +- .../org/apache/nifi/web/api/dto/UserDTO.java | 42 +- .../apache/nifi/web/api/dto/UserGroupDTO.java | 16 +- .../nifi/web/api/dto/action/ActionDTO.java | 40 +- .../nifi/web/api/dto/action/HistoryDTO.java | 12 +- .../web/api/dto/action/HistoryQueryDTO.java | 32 +- .../component/details/ExtensionDetailsDTO.java | 4 +- .../details/RemoteProcessGroupDetailsDTO.java | 4 +- .../dto/action/details/ConfigureDetailsDTO.java | 12 +- .../dto/action/details/ConnectDetailsDTO.java | 28 +- .../api/dto/action/details/MoveDetailsDTO.java | 16 +- .../api/dto/action/details/PurgeDetailsDTO.java | 4 +- .../web/api/dto/provenance/AttributeDTO.java | 12 +- .../web/api/dto/provenance/ProvenanceDTO.java | 36 +- .../api/dto/provenance/ProvenanceEventDTO.java | 186 +- .../dto/provenance/ProvenanceOptionsDTO.java | 4 +- .../dto/provenance/ProvenanceRequestDTO.java | 24 +- .../dto/provenance/ProvenanceResultsDTO.java | 28 +- .../ProvenanceSearchableFieldDTO.java | 16 +- .../api/dto/provenance/lineage/LineageDTO.java | 36 +- .../provenance/lineage/LineageRequestDTO.java | 12 +- .../provenance/lineage/LineageResultsDTO.java | 12 +- .../provenance/lineage/ProvenanceLinkDTO.java | 20 +- .../provenance/lineage/ProvenanceNodeDTO.java | 36 +- .../dto/search/ComponentSearchResultDTO.java | 16 +- .../web/api/dto/search/NodeSearchResultDTO.java | 8 +- .../web/api/dto/search/SearchResultsDTO.java | 28 +- .../dto/search/UserGroupSearchResultDTO.java | 4 +- .../web/api/dto/search/UserSearchResultDTO.java | 8 +- .../dto/status/ClusterConnectionStatusDTO.java | 14 +- .../api/dto/status/ClusterPortStatusDTO.java | 14 +- .../dto/status/ClusterProcessorStatusDTO.java | 20 +- .../ClusterRemoteProcessGroupStatusDTO.java | 14 +- .../web/api/dto/status/ClusterStatusDTO.java | 4 +- .../api/dto/status/ClusterStatusHistoryDTO.java | 12 +- .../web/api/dto/status/ConnectionStatusDTO.java | 42 +- .../web/api/dto/status/ControllerStatusDTO.java | 42 +- .../api/dto/status/NodeConnectionStatusDTO.java | 8 +- .../web/api/dto/status/NodePortStatusDTO.java | 8 +- .../api/dto/status/NodeProcessorStatusDTO.java | 8 +- .../status/NodeRemoteProcessGroupStatusDTO.java | 8 +- .../nifi/web/api/dto/status/NodeStatusDTO.java | 8 +- .../api/dto/status/NodeStatusHistoryDTO.java | 8 +- .../nifi/web/api/dto/status/PortStatusDTO.java | 36 +- .../api/dto/status/ProcessGroupStatusDTO.java | 8 +- .../web/api/dto/status/ProcessorStatusDTO.java | 46 +- .../web/api/dto/status/RemotePortStatusDTO.java | 20 +- .../dto/status/RemoteProcessGroupStatusDTO.java | 38 +- .../nifi/web/api/dto/status/StatusDTO.java | 4 +- .../web/api/dto/status/StatusDescriptorDTO.java | 16 +- .../web/api/dto/status/StatusHistoryDTO.java | 16 +- .../api/dto/status/StatusHistoryDetailDTO.java | 8 +- .../web/api/dto/status/StatusSnapshotDTO.java | 8 +- .../nifi/web/api/entity/AuthorityEntity.java | 8 +- .../api/entity/ClusterSearchResultsEntity.java | 4 +- .../nifi/web/api/entity/ConnectionEntity.java | 4 +- .../nifi/web/api/entity/ConnectionsEntity.java | 4 +- .../web/api/entity/ControllerServiceEntity.java | 6 +- ...ollerServiceReferencingComponentsEntity.java | 5 +- .../entity/ControllerServiceTypesEntity.java | 4 +- .../api/entity/ControllerServicesEntity.java | 4 +- .../nifi/web/api/entity/CounterEntity.java | 4 +- .../nifi/web/api/entity/CountersEntity.java | 6 +- .../org/apache/nifi/web/api/entity/Entity.java | 4 +- .../nifi/web/api/entity/FunnelsEntity.java | 4 +- .../nifi/web/api/entity/InputPortEntity.java | 4 +- .../nifi/web/api/entity/InputPortsEntity.java | 4 +- .../nifi/web/api/entity/LabelsEntity.java | 4 +- .../nifi/web/api/entity/OutputPortEntity.java | 4 +- .../nifi/web/api/entity/OutputPortsEntity.java | 4 +- .../web/api/entity/PrioritizerTypesEntity.java | 4 +- .../web/api/entity/ProcessorTypesEntity.java | 4 +- .../nifi/web/api/entity/ProcessorsEntity.java | 4 +- .../api/entity/PropertyDescriptorEntity.java | 1 - .../nifi/web/api/entity/ProvenanceEntity.java | 4 +- .../api/entity/RemoteProcessGroupsEntity.java | 4 +- .../web/api/entity/ReportingTaskEntity.java | 6 +- .../api/entity/ReportingTaskTypesEntity.java | 4 +- .../web/api/entity/ReportingTasksEntity.java | 4 +- .../web/api/entity/SearchResultsEntity.java | 4 +- .../nifi/web/api/entity/TemplatesEntity.java | 4 +- .../web/api/entity/UserSearchResultsEntity.java | 8 +- .../apache/nifi/web/api/entity/UsersEntity.java | 4 +- .../util/SslContextCreationException.java | 1 + .../security/util/SslContextFactory.java | 39 +- .../framework/security/util/SslException.java | 1 + .../security/util/SslServerSocketFactory.java | 19 +- ...SslServerSocketFactoryCreationException.java | 4 +- .../security/util/SslSocketFactory.java | 17 +- .../util/SslSocketFactoryCreationException.java | 1 + .../security/util/SslContextFactoryTest.java | 18 +- .../nifi-framework/nifi-site-to-site/pom.xml | 4 +- .../util/RemoteProcessGroupUtils.java | 120 +- .../nifi/remote/RemoteResourceFactory.java | 114 +- .../nifi/remote/RemoteResourceManager.java | 98 +- .../apache/nifi/remote/RemoteSiteListener.java | 2 +- .../nifi/remote/SocketRemoteSiteListener.java | 226 +-- .../nifi/remote/StandardRemoteGroupPort.java | 200 ++- .../nifi/remote/StandardRootGroupPort.java | 243 +-- .../exception/UnsupportedCodecException.java | 5 +- .../socket/ClusterManagerServerProtocol.java | 50 +- .../socket/SocketFlowFileServerProtocol.java | 254 +-- .../remote/TestStandardSiteToSiteProtocol.java | 28 +- .../io/socket/TestSocketChannelStreams.java | 62 +- .../io/socket/ssl/TestSSLSocketChannel.java | 114 +- .../java/org/apache/nifi/web/ContentAccess.java | 6 +- .../apache/nifi/web/ContentRequestContext.java | 24 +- .../apache/nifi/web/DownloadableContent.java | 12 +- .../cache/protocol/ProtocolHandshake.java | 37 +- .../protocol/exception/HandshakeException.java | 3 +- .../apache/nifi/update/attributes/Criteria.java | 24 +- .../update/attributes/serde/CriteriaSerDe.java | 8 +- 242 files changed, 4484 insertions(+), 5584 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/9faaef8c/nifi-parent/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-parent/pom.xml b/nifi-parent/pom.xml index fc1a19d..ba9f7ae 100644 --- a/nifi-parent/pom.xml +++ b/nifi-parent/pom.xml @@ -198,6 +198,13 @@ <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-checkstyle-plugin</artifactId> <version>2.15</version> + <dependencies> + <dependency> + <groupId>com.puppycrawl.tools</groupId> + <artifactId>checkstyle</artifactId> + <version>6.5</version> + </dependency> + </dependencies> </plugin> </plugins> </pluginManagement> @@ -310,7 +317,6 @@ <property name="allowSamelineMultipleAnnotations" value="true" /> </module> <module name="NonEmptyAtclauseDescription" /> - <module name="JavadocTagContinuationIndentation" /> <module name="JavadocMethod"> <property name="allowMissingJavadoc" value="true" /> <property name="allowMissingParamTags" value="true" /> @@ -325,15 +331,8 @@ </checkstyleRules> <violationSeverity>warning</violationSeverity> <includeTestSourceDirectory>true</includeTestSourceDirectory> - <excludes>**/HelpMojo.java</excludes> + <excludes>**/HelpMojo.java,**/generated-sources</excludes> </configuration> - <dependencies> - <dependency> - <groupId>com.puppycrawl.tools</groupId> - <artifactId>checkstyle</artifactId> - <version>6.3</version> - </dependency> - </dependencies> <executions> <execution> <id>check-style</id> @@ -379,7 +378,7 @@ <plugin> <groupId>org.apache.nifi</groupId> <artifactId>nifi-nar-maven-plugin</artifactId> - <version>1.0.0-incubating</version> + <version>1.0.1-incubating-SNAPSHOT</version> <extensions>true</extensions> </plugin> </plugins> http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/9faaef8c/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/BootstrapCodec.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/BootstrapCodec.java b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/BootstrapCodec.java index f03bf1e..3817875 100644 --- a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/BootstrapCodec.java +++ b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/BootstrapCodec.java @@ -27,66 +27,66 @@ import java.util.Arrays; import org.apache.nifi.bootstrap.exception.InvalidCommandException; - public class BootstrapCodec { - private final RunNiFi runner; - private final BufferedReader reader; - private final BufferedWriter writer; - - public BootstrapCodec(final RunNiFi runner, final InputStream in, final OutputStream out) { - this.runner = runner; - this.reader = new BufferedReader(new InputStreamReader(in)); - this.writer = new BufferedWriter(new OutputStreamWriter(out)); - } - - public void communicate() throws IOException { - final String line = reader.readLine(); - final String[] splits = line.split(" "); - if ( splits.length < 0 ) { - throw new IOException("Received invalid command from NiFi: " + line); - } - - final String cmd = splits[0]; - final String[] args; - if ( splits.length == 1 ) { - args = new String[0]; - } else { - args = Arrays.copyOfRange(splits, 1, splits.length); - } - - try { - processRequest(cmd, args); - } catch (final InvalidCommandException ice) { - throw new IOException("Received invalid command from NiFi: " + line + " : " + ice.getMessage() == null ? "" : "Details: " + ice.toString()); - } - } - private void processRequest(final String cmd, final String[] args) throws InvalidCommandException, IOException { - switch (cmd) { - case "PORT": { - if ( args.length != 2 ) { - throw new InvalidCommandException(); - } - - final int port; - try { - port = Integer.parseInt( args[0] ); - } catch (final NumberFormatException nfe) { - throw new InvalidCommandException("Invalid Port number; should be integer between 1 and 65535"); - } - - if ( port < 1 || port > 65535 ) { - throw new InvalidCommandException("Invalid Port number; should be integer between 1 and 65535"); - } + private final RunNiFi runner; + private final BufferedReader reader; + private final BufferedWriter writer; + + public BootstrapCodec(final RunNiFi runner, final InputStream in, final OutputStream out) { + this.runner = runner; + this.reader = new BufferedReader(new InputStreamReader(in)); + this.writer = new BufferedWriter(new OutputStreamWriter(out)); + } + + public void communicate() throws IOException { + final String line = reader.readLine(); + final String[] splits = line.split(" "); + if (splits.length < 0) { + throw new IOException("Received invalid command from NiFi: " + line); + } + + final String cmd = splits[0]; + final String[] args; + if (splits.length == 1) { + args = new String[0]; + } else { + args = Arrays.copyOfRange(splits, 1, splits.length); + } + + try { + processRequest(cmd, args); + } catch (final InvalidCommandException ice) { + throw new IOException("Received invalid command from NiFi: " + line + " : " + ice.getMessage() == null ? "" : "Details: " + ice.toString()); + } + } + + private void processRequest(final String cmd, final String[] args) throws InvalidCommandException, IOException { + switch (cmd) { + case "PORT": { + if (args.length != 2) { + throw new InvalidCommandException(); + } + + final int port; + try { + port = Integer.parseInt(args[0]); + } catch (final NumberFormatException nfe) { + throw new InvalidCommandException("Invalid Port number; should be integer between 1 and 65535"); + } + + if (port < 1 || port > 65535) { + throw new InvalidCommandException("Invalid Port number; should be integer between 1 and 65535"); + } + + final String secretKey = args[1]; - final String secretKey = args[1]; - - runner.setNiFiCommandControlPort(port, secretKey); - writer.write("OK"); - writer.newLine(); - writer.flush(); - } - break; - } - } + runner.setNiFiCommandControlPort(port, secretKey); + writer.write("OK"); + writer.newLine(); + writer.flush(); + } + break; + } + } } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/9faaef8c/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java index f05d45a..1713478 100644 --- a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java +++ b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java @@ -28,101 +28,103 @@ import java.util.concurrent.TimeUnit; import org.apache.nifi.bootstrap.util.LimitingInputStream; public class NiFiListener { - private ServerSocket serverSocket; - private volatile Listener listener; - - int start(final RunNiFi runner) throws IOException { - serverSocket = new ServerSocket(); - serverSocket.bind(new InetSocketAddress("localhost", 0)); - - final int localPort = serverSocket.getLocalPort(); - listener = new Listener(serverSocket, runner); - final Thread listenThread = new Thread(listener); - listenThread.setName("Listen to NiFi"); - listenThread.start(); - return localPort; - } - - public void stop() throws IOException { - final Listener listener = this.listener; - if ( listener == null ) { - return; - } - - listener.stop(); - } - - private class Listener implements Runnable { - private final ServerSocket serverSocket; - private final ExecutorService executor; - private final RunNiFi runner; - private volatile boolean stopped = false; - - public Listener(final ServerSocket serverSocket, final RunNiFi runner) { - this.serverSocket = serverSocket; - this.executor = Executors.newFixedThreadPool(2); - this.runner = runner; - } - - public void stop() throws IOException { - stopped = true; - - executor.shutdown(); - try { - executor.awaitTermination(3, TimeUnit.SECONDS); - } catch (final InterruptedException ie) { - } - - serverSocket.close(); - } - - @Override - public void run() { - while (!serverSocket.isClosed()) { - try { - if ( stopped ) { - return; - } - - final Socket socket; - try { - socket = serverSocket.accept(); - } catch (final IOException ioe) { - if ( stopped ) { - return; - } - - throw ioe; - } - - executor.submit(new Runnable() { - @Override - public void run() { - try { - // we want to ensure that we don't try to read data from an InputStream directly - // by a BufferedReader because any user on the system could open a socket and send - // a multi-gigabyte file without any new lines in order to crash the Bootstrap, - // which in turn may cause the Shutdown Hook to shutdown NiFi. - // So we will limit the amount of data to read to 4 KB - final InputStream limitingIn = new LimitingInputStream(socket.getInputStream(), 4096); - final BootstrapCodec codec = new BootstrapCodec(runner, limitingIn, socket.getOutputStream()); - codec.communicate(); - } catch (final Throwable t) { - System.out.println("Failed to communicate with NiFi due to " + t); - t.printStackTrace(); - } finally { - try { - socket.close(); - } catch (final IOException ioe) { - } - } - } - }); - } catch (final Throwable t) { - System.err.println("Failed to receive information from NiFi due to " + t); - t.printStackTrace(); - } - } - } - } + + private ServerSocket serverSocket; + private volatile Listener listener; + + int start(final RunNiFi runner) throws IOException { + serverSocket = new ServerSocket(); + serverSocket.bind(new InetSocketAddress("localhost", 0)); + + final int localPort = serverSocket.getLocalPort(); + listener = new Listener(serverSocket, runner); + final Thread listenThread = new Thread(listener); + listenThread.setName("Listen to NiFi"); + listenThread.start(); + return localPort; + } + + public void stop() throws IOException { + final Listener listener = this.listener; + if (listener == null) { + return; + } + + listener.stop(); + } + + private class Listener implements Runnable { + + private final ServerSocket serverSocket; + private final ExecutorService executor; + private final RunNiFi runner; + private volatile boolean stopped = false; + + public Listener(final ServerSocket serverSocket, final RunNiFi runner) { + this.serverSocket = serverSocket; + this.executor = Executors.newFixedThreadPool(2); + this.runner = runner; + } + + public void stop() throws IOException { + stopped = true; + + executor.shutdown(); + try { + executor.awaitTermination(3, TimeUnit.SECONDS); + } catch (final InterruptedException ie) { + } + + serverSocket.close(); + } + + @Override + public void run() { + while (!serverSocket.isClosed()) { + try { + if (stopped) { + return; + } + + final Socket socket; + try { + socket = serverSocket.accept(); + } catch (final IOException ioe) { + if (stopped) { + return; + } + + throw ioe; + } + + executor.submit(new Runnable() { + @Override + public void run() { + try { + // we want to ensure that we don't try to read data from an InputStream directly + // by a BufferedReader because any user on the system could open a socket and send + // a multi-gigabyte file without any new lines in order to crash the Bootstrap, + // which in turn may cause the Shutdown Hook to shutdown NiFi. + // So we will limit the amount of data to read to 4 KB + final InputStream limitingIn = new LimitingInputStream(socket.getInputStream(), 4096); + final BootstrapCodec codec = new BootstrapCodec(runner, limitingIn, socket.getOutputStream()); + codec.communicate(); + } catch (final Throwable t) { + System.out.println("Failed to communicate with NiFi due to " + t); + t.printStackTrace(); + } finally { + try { + socket.close(); + } catch (final IOException ioe) { + } + } + } + }); + } catch (final Throwable t) { + System.err.println("Failed to receive information from NiFi due to " + t); + t.printStackTrace(); + } + } + } + } }
