in plugins/transforms/tokenreplacement/src/main/java/org/apache/hop/pipeline/transforms/tokenreplacement/TokenReplacement.java [63:287]
public synchronized boolean processRow() throws HopException {
boolean result = true;
Object[] r = getRow(); // This also waits for a row to be finished.
if (first && r != null) {
first = false;
data.inputRowMeta = getInputRowMeta();
data.outputRowMeta = getInputRowMeta().clone();
if (meta.getOutputType().equalsIgnoreCase(CONST_FIELD)) {
meta.getFields(data.outputRowMeta, getTransformName(), null, null, this, metadataProvider);
}
if (meta.getOutputType().equalsIgnoreCase("file") && !meta.isOutputFileNameInField()) {
if (meta.getOutputFileName() != null) {
String filename =
meta.buildFilename(
meta.getOutputFileName(), this, getCopy(), getPartitionId(), data.splitnr);
openNewOutputFile(filename);
} else {
throw new HopException("Output file name cannot be null.");
}
}
}
if (r == null) {
// no more input to be expected...
closeAllOutputFiles();
setOutputDone();
return false;
}
if (meta.getOutputType().equalsIgnoreCase("file")
&& !meta.isOutputFileNameInField()
&& meta.getSplitEvery() > 0
&& data.rowNumber % meta.getSplitEvery() == 0
&& data.rowNumber > 0) {
closeAllOutputFiles();
data.splitnr++;
String filename =
meta.buildFilename(
meta.getOutputFileName(), this, getCopy(), getPartitionId(), data.splitnr);
openNewOutputFile(filename);
}
String outputFilename = "";
if (meta.getOutputType().equalsIgnoreCase("file") && !meta.isOutputFileNameInField()) {
outputFilename =
meta.buildFilename(
meta.getOutputFileName(), this, getCopy(), getPartitionId(), data.splitnr);
} else if (meta.getOutputType().equalsIgnoreCase("file") && meta.isOutputFileNameInField()) {
String filenameValue =
data.inputRowMeta.getString(r, resolve(meta.getOutputFileNameField()), "");
if (!Utils.isEmpty(filenameValue)) {
outputFilename = filenameValue;
} else {
throw new HopException("Filename cannot be empty.");
}
}
// Create token resolver
TokenResolver resolver = new TokenResolver();
for (TokenReplacementField field : meta.getTokenReplacementFields()) {
if (data.inputRowMeta.indexOfValue(field.getName()) >= 0) {
String fieldValue = resolve(data.inputRowMeta.getString(r, field.getName(), null));
if (fieldValue == null
&& !BooleanUtils.toBoolean(
Const.getEnvironmentVariable("KETTLE_EMPTY_STRING_DIFFERS_FROM_NULL", "N"))) {
fieldValue = Const.nullToEmpty(fieldValue);
}
resolver.addToken(field.getTokenName(), fieldValue);
} else {
throw new HopValueException("Field " + field.getName() + CONST_NOT_FOUND_ON_INPUT_STREAM);
}
}
Reader reader;
String inputFilename = "";
if (meta.getInputType().equalsIgnoreCase("text")) {
reader =
new TokenReplacingReader(
resolver,
new StringReader(meta.getInputText()),
resolve(meta.getTokenStartString()),
resolve(meta.getTokenEndString()));
} else if (meta.getInputType().equalsIgnoreCase(CONST_FIELD)) {
if (data.inputRowMeta.indexOfValue(meta.getInputFieldName()) >= 0) {
String inputString = data.inputRowMeta.getString(r, meta.getInputFieldName(), "");
reader =
new TokenReplacingReader(
resolver,
new StringReader(inputString),
resolve(meta.getTokenStartString()),
resolve(meta.getTokenEndString()));
} else {
throw new HopValueException(
"Input field " + meta.getInputFieldName() + CONST_NOT_FOUND_ON_INPUT_STREAM);
}
} else if (meta.getInputType().equalsIgnoreCase("file")) {
if (meta.isInputFileNameInField()) {
if (data.inputRowMeta.indexOfValue(resolve(meta.getInputFileNameField())) >= 0) {
inputFilename = data.inputRowMeta.getString(r, resolve(meta.getInputFileNameField()), "");
} else {
throw new HopValueException(
"Input filename field "
+ resolve(meta.getInputFileNameField())
+ CONST_NOT_FOUND_ON_INPUT_STREAM);
}
} else {
inputFilename = resolve(meta.getInputFileName());
}
if (Utils.isEmpty(inputFilename)) {
throw new HopValueException("Input filename cannot be empty");
}
if (!HopVfs.fileExists(inputFilename, variables)) {
throw new HopException("Input file " + inputFilename + " does not exist.");
}
reader =
new TokenReplacingReader(
resolver,
new InputStreamReader(HopVfs.getInputStream(inputFilename, variables)),
resolve(meta.getTokenStartString()),
resolve(meta.getTokenEndString()));
if (meta.isAddInputFileNameToResult()) {
ResultFile resultFile =
new ResultFile(
ResultFile.FILE_TYPE_GENERAL,
HopVfs.getFileObject(inputFilename, variables),
getTransformMeta().getName(),
getTransformName());
resultFile.setComment(BaseMessages.getString(PKG, "TokenReplacement.AddInputResultFile"));
addResultFile(resultFile);
}
} else {
throw new HopException("Unsupported input type " + meta.getInputType());
}
Writer stringWriter = null;
OutputStream bufferedWriter = null;
if (meta.getOutputType().equalsIgnoreCase(CONST_FIELD)) {
stringWriter = new StringWriter(5000);
} else {
if (meta.getOutputType().equalsIgnoreCase("file")) {
if (inputFilename.equals(outputFilename)) {
throw new HopException(
"Input and output filenames must not be the same " + inputFilename);
}
int fileIndex = data.openFiles.indexOf(outputFilename);
if (fileIndex < 0) {
openNewOutputFile(outputFilename);
fileIndex = data.openFiles.indexOf(outputFilename);
}
bufferedWriter = data.openBufferedWriters.get(fileIndex);
} else {
throw new HopException("Unsupported output type " + meta.getOutputType());
}
}
String output = "";
try {
char[] cbuf = new char[5000];
int length = 0;
while ((length = reader.read(cbuf)) > 0) {
if (meta.getOutputType().equalsIgnoreCase(CONST_FIELD)) {
stringWriter.write(cbuf, 0, length);
} else if (meta.getOutputType().equalsIgnoreCase("file")) {
CharBuffer cBuffer = CharBuffer.wrap(cbuf, 0, length);
ByteBuffer bBuffer = Charset.forName(meta.getOutputFileEncoding()).encode(cBuffer);
byte[] bytes = new byte[bBuffer.limit()];
bBuffer.get(bytes);
bufferedWriter.write(bytes);
} // No else. Anything else will be thrown to a Hop exception prior to getting here.
cbuf = new char[5000];
}
if (meta.getOutputType().equalsIgnoreCase(CONST_FIELD)) {
output += stringWriter.toString();
} else if (meta.getOutputType().equalsIgnoreCase("file")) {
bufferedWriter.write(meta.getOutputFileFormatString().getBytes());
}
} catch (IOException ex) {
throw new HopException(ex.getMessage(), ex);
} finally {
try {
reader.close();
if (stringWriter != null) {
stringWriter.close();
}
reader = null;
stringWriter = null;
} catch (IOException ex) {
throw new HopException(ex.getMessage(), ex);
}
}
if (meta.getOutputType().equalsIgnoreCase(CONST_FIELD)) {
r = RowDataUtil.addValueData(r, data.outputRowMeta.size() - 1, output);
} else if (meta.getOutputType().equalsIgnoreCase("file")) {
incrementLinesWritten();
}
putRow(data.outputRowMeta, r); // in case we want it to go further...
data.rowNumber++;
if (checkFeedback(getLinesOutput())) {
logBasic("linenr " + getLinesOutput());
}
return result;
}