2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2016-2018 Ericsson. All rights reserved.
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.apex.service.engine.event.impl.filecarrierplugin.consumer;
23 import java.io.BufferedReader;
24 import java.io.IOException;
25 import java.io.InputStream;
26 import java.io.InputStreamReader;
27 import java.util.Queue;
28 import java.util.concurrent.LinkedBlockingQueue;
30 import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
31 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
32 import org.onap.policy.apex.service.parameters.eventprotocol.EventProtocolTextTokenDelimitedParameters;
33 import org.slf4j.ext.XLogger;
34 import org.slf4j.ext.XLoggerFactory;
37 * The Class TextBlockReader reads the next block of text from an input stream.
39 * @author Liam Fallon (liam.fallon@ericsson.com)
41 public class HeaderDelimitedTextBlockReader implements TextBlockReader, Runnable {
42 // The logger for this class
43 private static final XLogger LOGGER = XLoggerFactory.getXLogger(HeaderDelimitedTextBlockReader.class);
45 // The amount of time to wait for input on the text block reader
46 private static final long TEXT_BLOCK_DELAY = 250;
48 // Tag for the start and end of text blocks
49 private final String blockStartToken;
50 private final String blockEndToken;
52 // Indicates that text block processing starts at the first block of text
53 private final boolean delimiterAtStart;
54 private boolean blockEndTokenUsed = false;
56 // The thread used to read the text from the stream
57 Thread textConsumputionThread;
59 // The input stream for text
60 private InputStream inputStream;
62 // The lines of input read from the input stream
63 private final Queue<String> textLineQueue = new LinkedBlockingQueue<>();
65 // True while EOF has not been seen on input
66 private boolean eofOnInputStream = false;
69 * Constructor, initialize the text block reader using token delimited event protocol parameters.
71 * @param tokenDelimitedParameters
72 * the token delimited event protocol parameters
74 public HeaderDelimitedTextBlockReader(final EventProtocolTextTokenDelimitedParameters tokenDelimitedParameters) {
75 this(tokenDelimitedParameters.getStartDelimiterToken(), tokenDelimitedParameters.getEndDelimiterToken(),
76 tokenDelimitedParameters.isDelimiterAtStart());
80 * Constructor, initialize the text block reader.
82 * @param blockStartToken
83 * the block start token for the start of a text block
84 * @param blockEndToken
85 * the block end token for the end of a text block
86 * @param delimiterAtStart
87 * indicates that text block processing starts at the first block of text
89 public HeaderDelimitedTextBlockReader(final String blockStartToken, final String blockEndToken,
90 final boolean delimiterAtStart) {
91 this.blockStartToken = blockStartToken;
92 this.delimiterAtStart = delimiterAtStart;
94 if (blockEndToken == null) {
95 this.blockEndToken = blockStartToken;
96 this.blockEndTokenUsed = false;
98 this.blockEndToken = blockEndToken;
99 this.blockEndTokenUsed = true;
106 * @see org.onap.policy.apex.service.engine.event.impl.filecarrierplugin.consumer.TextBlockReader# init(
107 * java.io.InputStream)
110 public void init(final InputStream incomingInputStream) {
111 this.inputStream = incomingInputStream;
113 // Configure and start the text reading thread
114 textConsumputionThread = new ApplicationThreadFactory(this.getClass().getName()).newThread(this);
115 textConsumputionThread.setDaemon(true);
116 textConsumputionThread.start();
122 * @see org.onap.policy.apex.service.engine.event.impl.filecarrierplugin.consumer.TextBlockReader# readTextBlock()
125 public TextBlock readTextBlock() throws IOException {
126 // Holder for the current text block
127 final StringBuilder textBlockBuilder = new StringBuilder();
129 // Wait for the timeout period if there is no input
130 if (!eofOnInputStream && textLineQueue.isEmpty()) {
131 ThreadUtilities.sleep(TEXT_BLOCK_DELAY);
134 // Scan the lines in the queue
135 while (!textLineQueue.isEmpty()) {
136 // Scroll down in the available lines looking for the start of the text block
137 if (!delimiterAtStart || textLineQueue.peek().startsWith(blockStartToken)) {
138 // Process the input line header
139 textBlockBuilder.append(textLineQueue.remove());
140 textBlockBuilder.append('\n');
143 String consumer = textLineQueue.remove();
144 LOGGER.warn("invalid input on consumer: {}", consumer);
148 // Get the rest of the text document
149 while (!textLineQueue.isEmpty() && !textLineQueue.peek().startsWith(blockEndToken)
150 && !textLineQueue.peek().startsWith(blockStartToken)) {
151 // We just strip out block end tokens because we use block start tokens to delimit the blocks of text
152 textBlockBuilder.append(textLineQueue.remove());
153 textBlockBuilder.append('\n');
156 // Check if we should add the block end token to the end of the text block
157 if (!textLineQueue.isEmpty() && blockEndTokenUsed && textLineQueue.peek().startsWith(blockEndToken)) {
158 // Process the input line header
159 textBlockBuilder.append(textLineQueue.remove());
160 textBlockBuilder.append('\n');
163 // Condition the text block and return it
164 final String textBlock = textBlockBuilder.toString().trim();
165 final boolean endOfText = eofOnInputStream && textLineQueue.isEmpty();
167 if (textBlock.length() > 0) {
168 return new TextBlock(endOfText, textBlock);
170 return new TextBlock(endOfText, null);
177 * @see java.lang.Runnable#run()
181 final BufferedReader textReader = new BufferedReader(new InputStreamReader(inputStream));
184 // Read the input line by line until we see end of file on the stream
186 while ((line = textReader.readLine()) != null) {
187 textLineQueue.add(line);
189 } catch (final IOException e) {
190 LOGGER.warn("I/O exception on text input on consumer: ", e);
192 eofOnInputStream = true;