[ https://issues.apache.org/jira/browse/NIFI-3518?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15890485#comment-15890485 ]
ASF GitHub Bot commented on NIFI-3518: -------------------------------------- Github user WilliamNouet commented on a diff in the pull request: https://github.com/apache/nifi/pull/1529#discussion_r103725860 --- Diff: nifi-nar-bundles/nifi-morphlines-bundle/nifi-morphlines-processor/src/main/java/org/apache/nifi/processors/morphlines/Morphlines.java --- @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.morphlines; + +import com.google.common.base.Preconditions; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.*; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.InputStreamCallback; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.io.StreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.stream.io.StreamUtils; +import org.kitesdk.morphline.api.Command; +import org.kitesdk.morphline.api.MorphlineContext; +import org.kitesdk.morphline.api.Record; +import org.kitesdk.morphline.base.Fields; + +import org.kitesdk.morphline.base.Notifications; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import org.apache.nifi.annotation.lifecycle.OnScheduled; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.*; +import java.util.stream.*; +import java.util.concurrent.atomic.*; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.processor.exception.*; + +@Tags({"kitesdk", "morphlines"}) +@CapabilityDescription("Implements Morphlines framework") +public class MorphlinesProcessor extends AbstractProcessor { + + private static Command morphline; + private volatile Record record = new Record(); + private static volatile Collector collector = new Collector(); + + public static final PropertyDescriptor MORPHLINES_ID = new PropertyDescriptor + .Builder().name("Morphlines ID") + .description("Identifier of the morphlines context") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + + public static final PropertyDescriptor MORPHLINES_FILE = new PropertyDescriptor + .Builder().name("Morphlines File") + .description("File for the morphlines context") + .required(true) + .addValidator(StandardValidators.FILE_EXISTS_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + + public static final PropertyDescriptor MORPHLINES_OUTPUT_FIELD = new PropertyDescriptor + .Builder().name("Morphlines output field") + .description("Field name of output in Morphlines. Default is '_attachment_body'.") + .required(false) + .defaultValue("_attachment_body") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final Relationship REL_SUCCESS = new Relationship.Builder() + .name("success") + .description("Relationship for success.") + .build(); + + public static final Relationship REL_FAILURE = new Relationship.Builder() + .name("failure") + .description("Relationship for failure of morphlines.") + .build(); + + public static final Relationship REL_ORIGINAL = new Relationship.Builder() + .name("original") + .description("Relationship for original flowfiles.") + .build(); + + private static final List<PropertyDescriptor> PROPERTIES = ImmutableList.<PropertyDescriptor>builder() + .add(MORPHLINES_FILE) + .add(MORPHLINES_ID) + .add(MORPHLINES_OUTPUT_FIELD) + .build(); + + private static final Set<Relationship> RELATIONSHIPS = ImmutableSet.<Relationship>builder() + .add(REL_SUCCESS) + .add(REL_FAILURE) + .build(); + + private File morphLineFile; + private String morphLinesId; + private String morphlinesOutputField; + private MorphlineContext morphlineContext; + + @Override + public Set<Relationship> getRelationships() { + return RELATIONSHIPS; + } + + @Override + public final List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTIES; + } + + @OnScheduled + public void onScheduled(ProcessContext context) throws Exception { + morphLineFile = new File(context.getProperty(MORPHLINES_FILE).getValue()); + morphLinesId = context.getProperty(MORPHLINES_ID).getValue(); + morphlinesOutputField = context.getProperty(MORPHLINES_OUTPUT_FIELD).getValue(); + morphlineContext = new MorphlineContext.Builder().build(); + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + FlowFile flowFile = session.get(); + if ( flowFile == null ) { + return; + } + + final AtomicLong written = new AtomicLong(0L); + final byte[] value = new byte[(int) flowFile.getSize()]; + + try{ + flowFile = session.write(flowFile, new StreamCallback() { + @Override + public void process(InputStream in, OutputStream out) throws IOException { + StreamUtils.fillBuffer(in, value); + Record record = new Record(); + record.put(Fields.ATTACHMENT_BODY, value); + Collector collectorRecord = new Collector(); + morphline = new org.kitesdk.morphline.base.Compiler().compile(morphLineFile, morphLinesId, morphlineContext, collectorRecord); --- End diff -- Well no because the collectorRecord is specific to a given flowfile, and as such, needs to be re-initialized for every flowfile and compile the Morphline command on the onTrigger method. This is not inefficient since the Morphlines processing happens in morphline.process(record). > Create a Morphlines processor > ----------------------------- > > Key: NIFI-3518 > URL: https://issues.apache.org/jira/browse/NIFI-3518 > Project: Apache NiFi > Issue Type: New Feature > Reporter: William Nouet > Priority: Minor > > Create a dedicate processor to run Morphlines transformations > (http://kitesdk.org/docs/1.1.0/morphlines/morphlines-reference-guide.html) -- This message was sent by Atlassian JIRA (v6.3.15#6346)