Below is my custom processor which selects data from mongoDB and write it in flowFile stream. Code is working fine but I need to modify the connection part so it uses connection pool instead of traditional connection. Similar to GetMongo processor String db_name = "cms", host_name = "localhost"; MongoClient mongo_client = new MongoClient(host_name, 27017); MongoDatabase db = mongo_client.getDatabase(db_name); How to achieve this, Custom processor Below /* * 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 test.processors.test2; import com.mongodb.MongoClient; import com.mongodb.client.FindIterable; import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoCursor; import com.mongodb.client.MongoDatabase; import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.InputStream; import java.io.InputStreamReader; import java.io.OutputStream; import java.nio.charset.StandardCharsets; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.flowfile.FlowFile; import org.apache.nifi.annotation.behavior.ReadsAttribute; import org.apache.nifi.annotation.behavior.ReadsAttributes; import org.apache.nifi.annotation.behavior.WritesAttribute; import org.apache.nifi.annotation.behavior.WritesAttributes; import org.apache.nifi.annotation.lifecycle.OnScheduled; import org.apache.nifi.annotation.documentation.CapabilityDescription; import org.apache.nifi.annotation.documentation.SeeAlso; import org.apache.nifi.annotation.documentation.Tags; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.AbstractProcessor; import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.util.StandardValidators; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import java.io.OutputStreamWriter; import org.apache.nifi.logging.ComponentLog; import org.apache.nifi.processor.exception.FlowFileAccessException; import org.bson.Document; @Tags({"example"}) @CapabilityDescription("Provide a description") @SeeAlso({}) @ReadsAttributes({ @ReadsAttribute(attribute = "", description = "")}) @WritesAttributes({ @WritesAttribute(attribute = "", description = "")}) public class MyProcessor extends AbstractProcessor { public static final PropertyDescriptor MY_PROPERTY = new PropertyDescriptor.Builder().name("MY_PROPERTY") .displayName("Prefix property") .description("Prefix Property") .required(true) .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) .build(); public static final Relationship REL_SUCCESS = new Relationship.Builder() .name("Success") .description("Sucessful relationship") .build(); public static final Relationship REL_FAILURE = new Relationship.Builder() .name("Fail") .description("Failure relationship") .build(); private List<PropertyDescriptor> descriptors; private Set<Relationship> relationships; @Override protected void init(final ProcessorInitializationContext context) { final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>(); descriptors.add(MY_PROPERTY); this.descriptors = Collections.unmodifiableList(descriptors); final Set<Relationship> relationships = new HashSet<Relationship>(); relationships.add(REL_SUCCESS); relationships.add(REL_FAILURE); this.relationships = Collections.unmodifiableSet(relationships); } @Override public Set<Relationship> getRelationships() { return this.relationships; } @Override public final List<PropertyDescriptor> getSupportedPropertyDescriptors() { return descriptors; } @OnScheduled public void onScheduled(final ProcessContext context) { } @Override public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { FlowFile flowFile = session.get(); if (flowFile == null) { return; } final ComponentLog logger = getLogger(); final String prefixStr = context.getProperty(MY_PROPERTY).getValue(); try { flowFile = session.write(flowFile, (final InputStream inputStream, final OutputStream outputStream) -> { BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(outputStream, StandardCharsets.UTF_8)); String line; while ((line = bufferedReader.readLine()) != null) { String xxx = mongoDB( prefixStr, line); String aggregatedText = prefixStr + ": " + line + ": " + xxx; bufferedWriter.write(aggregatedText); bufferedWriter.newLine(); } bufferedWriter.flush(); }); session.transfer(flowFile, REL_SUCCESS); logger.info("successfully processed FlowFile {}", new Object[]{flowFile}); } catch (FlowFileAccessException e) { session.transfer(flowFile, REL_FAILURE); logger.info("Failed to add prefix for FlowFile: " + e.getMessage()); } } String db_name = "cms", host_name = "localhost"; MongoClient mongo_client = new MongoClient(host_name, 27017); MongoDatabase db = mongo_client.getDatabase(db_name); public String mongoDB(String arg1, String arg2) { final ComponentLog logger = getLogger(); String result = ""; try { String db_col_name = arg1; MongoCollection<Document> coll = db.getCollection(db_col_name); Document query = new Document(); query.put("msisdn", arg2); logger.info(query.toJson()); FindIterable<Document> fi = coll.find(query); try (MongoCursor<Document> cursor = fi.iterator()) { while (cursor.hasNext()) { result = result + cursor.next().toJson(); } } } catch (Exception e) { logger.info("Error: " + e.getMessage()); } return result; } } ... View more

Hi All, I'm trying to create simple custom processor, I used to work with executeScript processor to for loop using the below flowFile = session.write(flowFile, {inputStream, outputStream -> inputStream.eachLine { line -> def a = line.split("\\|", -1) outputStream.write("$Test ABCABC\n".toString().getBytes(StandardCharsets.UTF_8)) } } as StreamCallback) How can I rewrite this in Java, to inject in the below custom processor code /* * 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 hwx.processors.hwx; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.flowfile.FlowFile; import org.apache.nifi.annotation.behavior.ReadsAttribute; import org.apache.nifi.annotation.behavior.ReadsAttributes; import org.apache.nifi.annotation.behavior.WritesAttribute; import org.apache.nifi.annotation.behavior.WritesAttributes; import org.apache.nifi.annotation.lifecycle.OnScheduled; import org.apache.nifi.annotation.documentation.CapabilityDescription; import org.apache.nifi.annotation.documentation.SeeAlso; import org.apache.nifi.annotation.documentation.Tags; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.AbstractProcessor; import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.util.StandardValidators; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; @Tags({"example"}) @CapabilityDescription("Provide a description") @SeeAlso({}) @ReadsAttributes({@ReadsAttribute(attribute="", description="")}) @WritesAttributes({@WritesAttribute(attribute="", description="")}) public class MyProcessor extends AbstractProcessor { public static final PropertyDescriptor MY_PROPERTY = new PropertyDescriptor .Builder().name("MY_PROPERTY") .displayName("My property") .description("Example Property") .required(true) .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) .build(); public static final Relationship MY_RELATIONSHIP = new Relationship.Builder() .name("MY_RELATIONSHIP") .description("Example relationship") .build(); private List<PropertyDescriptor> descriptors; private Set<Relationship> relationships; @Override protected void init(final ProcessorInitializationContext context) { final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>(); descriptors.add(MY_PROPERTY); this.descriptors = Collections.unmodifiableList(descriptors); final Set<Relationship> relationships = new HashSet<Relationship>(); relationships.add(MY_RELATIONSHIP); this.relationships = Collections.unmodifiableSet(relationships); } @Override public Set<Relationship> getRelationships() { return this.relationships; } @Override public final List<PropertyDescriptor> getSupportedPropertyDescriptors() { return descriptors; } @OnScheduled public void onScheduled(final ProcessContext context) { } @Override public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { FlowFile flowFile = session.get(); if ( flowFile == null ) { return; } ////// Code here ///////////////// session.transfer(flowFile, MY_RELATIONSHIP); } } Thanks a lot ... View more

Dears I managed to create simple InvokeScriptedProcessor.. Code is working fine. However, if I made code error intentionally I get the below error, UndeclaredThrowableException. 2018-12-11 12:30:59,400 WARN [Timer-Driven Process Thread-7] o.a.n.c.t.ContinuallyRunProcessorTask Administratively Yielding InvokeScriptedProcessor[id=115d1249-1fba-1c84-7af8-8918fa3e1ecd] due to uncaught Exception: java.lang.reflect.UndeclaredThrowableException 2018-12-11 12:30:59,400 WARN [Timer-Driven Process Thread-7] o.a.n.c.t.ContinuallyRunProcessorTask java.lang.reflect.UndeclaredThrowableException: null Code: import java.nio.charset.StandardCharsets import java.io.IOException class TESTSCRIPT implements Processor { def REL_SUCCESS = new Relationship.Builder() .name("success") .description("The flowfile with the specified query results was successfully transferred.") .build(); def REL_FAILURE = new Relationship.Builder() .name("failure") .description("An error occured while running the specified query.") .build(); def REL_BACK = new Relationship.Builder() .name("backup") .description("Backup files.") .build(); def PORT_NUMBER = new PropertyDescriptor.Builder() .name('Port number') .description('Port number') .required(true) .expressionLanguageSupported(true) .addValidator(StandardValidators.PORT_VALIDATOR) .build() def FILENAME = new PropertyDescriptor.Builder() .name('File Name') .description('Sets the filename attribute of the flowfile (do not include the extension).') .required(true) .expressionLanguageSupported(true) .addValidator(Validator.VALID) .build() def ComponentLog log @Override void initialize(ProcessorInitializationContext context) { log = context.getLogger() } @Override Set<Relationship> getRelationships() { return [REL_SUCCESS, REL_FAILURE, REL_BACK] as Set } @Override void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { ///// Code def session = sessionFactory.createSession() def flowFile = session.get() if(!flowFile) return def FILENAME = context.getProperty(FILENAME).value def PORT_NUMBER = context.getProperty(PORT_NUMBER).value flowFile = session.write(flowFile, {inputStream, outputStream -> inputStream.eachLine { line -> def a = line.split("\\|", -1) outputStream.write("${PORT_NUMBER}: Test\n".toString().getBytes(StandardCharsets.UTF_8)) } } as StreamCallback) flowFile = session.putAttribute(flowFile, "filename", FILENAME) try { FlowFile flowFilec = session.clone(flowFile); //log.error("ExecuteScript: transfer file") session.transfer(flowFile, REL_SUCCESS) session.transfer(flowFilec, REL_BACK) session.commit() } catch (e) { log.error('Scripting error: ', e) session.transfer(flowFile, REL_FAILURE) //session.rollback(true) session.commit() } } //////////// @Override Collection<ValidationResult> validate(ValidationContext context) { return null } @Override PropertyDescriptor getPropertyDescriptor(String name) { switch(name) { case 'File Name': return FILENAME case 'SMPP Port number': return PORT_NUMBER default: return null } } @Override void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue) { } @Override List<PropertyDescriptor> getPropertyDescriptors() { return [FILENAME, PORT_NUMBER] as List } @Override String getIdentifier() { return 'TESTSCRIPT-InvokeScriptedProcessor' } } processor = new TESTSCRIPT() Not sure why this Exception is generated, how can I handle it? so I get only the errors that are related to the code problem ... View more

Hello forum, I have Java code and I want to include it within the ExecuteScript processor, This is the Java code, import java.io.IOException import java.util.Date import java.util.concurrent.ExecutorService import java.util.concurrent.Executors import java.util.concurrent.TimeoutException import org.apache.log4j.BasicConfigurator import org.jsmpp.PDUStringException import org.jsmpp.SMPPConstant import org.jsmpp.bean.CancelSm import org.jsmpp.bean.DataCoding import org.jsmpp.bean.DataSm import org.jsmpp.bean.DeliveryReceipt import org.jsmpp.bean.ESMClass import org.jsmpp.bean.GSMSpecificFeature import org.jsmpp.bean.MessageMode import org.jsmpp.bean.MessageType import org.jsmpp.bean.NumberingPlanIndicator import org.jsmpp.bean.QuerySm import org.jsmpp.bean.RegisteredDelivery import org.jsmpp.bean.ReplaceSm import org.jsmpp.bean.SMSCDeliveryReceipt import org.jsmpp.bean.SubmitMulti import org.jsmpp.bean.SubmitMultiResult import org.jsmpp.bean.SubmitSm import org.jsmpp.bean.TypeOfNumber import org.jsmpp.extra.ProcessRequestException import org.jsmpp.session.BindRequest import org.jsmpp.session.DataSmResult import org.jsmpp.session.QuerySmResult import org.jsmpp.session.SMPPServerSession import org.jsmpp.session.SMPPServerSessionListener import org.jsmpp.session.ServerMessageReceiverListener import org.jsmpp.session.ServerResponseDeliveryAdapter import org.jsmpp.session.Session import org.jsmpp.util.DeliveryReceiptState import org.jsmpp.util.MessageIDGenerator import org.jsmpp.util.MessageId import org.jsmpp.util.RandomMessageIDGenerator import org.slf4j.Logger import org.slf4j.LoggerFactory import org.jsmpp.* import jsmpp.* public class SMPPServerSimulator extends ServerResponseDeliveryAdapter implements Runnable, ServerMessageReceiverListener { private static final Integer DEFAULT_PORT = 8057 private static final Logger logger = LoggerFactory.getLogger(SMPPServerSimulator.class) private final ExecutorService execService = Executors.newFixedThreadPool(5) private final ExecutorService execServiceDelReciept = Executors.newFixedThreadPool(100) private final MessageIDGenerator messageIDGenerator = new RandomMessageIDGenerator() private int port public SMPPServerSimulator(int port) { this.port = port } public void run() { try { SMPPServerSessionListener sessionListener = new SMPPServerSessionListener(port) logger.info("Listening on port {}", port) while (true) { SMPPServerSession serverSession = sessionListener.accept() logger.info("Accepting connection for session {}", serverSession.getSessionId()) serverSession.setMessageReceiverListener(this) serverSession.setResponseDeliveryListener(this) execService.execute(new WaitBindTask(serverSession)) } } catch (IOException e) { logger.error("IO error occured", e) } } @Override public QuerySmResult onAcceptQuerySm(QuerySm querySm, SMPPServerSession source) throws ProcessRequestException { logger.info("Accepting query sm, but not implemented") return null } @Override public MessageId onAcceptSubmitSm(SubmitSm submitSm, SMPPServerSession source) throws ProcessRequestException { MessageId messageId = messageIDGenerator.newMessageId() logger.debug("\nReceiving submit_sm {}, and return message id {}\n", new String(submitSm.getShortMessage()), messageId.getValue()) if (SMSCDeliveryReceipt.SUCCESS.containedIn(submitSm.getRegisteredDelivery()) || SMSCDeliveryReceipt.SUCCESS_FAILURE.containedIn(submitSm.getRegisteredDelivery())) { execServiceDelReciept.execute(new DeliveryReceiptTask(source, submitSm, messageId)) } flowFile = session.create() flowFile = session.putAttribute(flowFile, 'myAttr', new String(submitSm.getShortMessage())) session.transfer(flowFile, REL_SUCCESS) System.out.println("Message Recieved: " + new String(submitSm.getShortMessage())) return messageId } @Override public void onSubmitSmRespSent(MessageId messageId, SMPPServerSession source) { logger.debug("submit_sm_resp with message_id {} has been sent", messageId) } @Override public SubmitMultiResult onAcceptSubmitMulti(SubmitMulti submitMulti, SMPPServerSession source) throws ProcessRequestException { return null } @Override public DataSmResult onAcceptDataSm(DataSm dataSm, Session source) throws ProcessRequestException { return null } @Override public void onAcceptCancelSm(CancelSm cancelSm, SMPPServerSession source) throws ProcessRequestException { } @Override public void onAcceptReplaceSm(ReplaceSm replaceSm, SMPPServerSession source) throws ProcessRequestException { } private class WaitBindTask implements Runnable { private final SMPPServerSession serverSession public WaitBindTask(SMPPServerSession serverSession) { this.serverSession = serverSession } @Override public void run() { try { BindRequest bindRequest = serverSession.waitForBind(1000) logger.info("Accepting bind for session {}", serverSession.getSessionId()) try { bindRequest.accept("sys") } catch (PDUStringException e) { logger.error("Invalid system id", e) bindRequest.reject(SMPPConstant.STAT_ESME_RSYSERR) } } catch (IllegalStateException e) { logger.error("System error", e) } catch (TimeoutException e) { logger.warn("Wait for bind has reach timeout", e) } catch (IOException e) { logger.error("Failed accepting bind request for session {}", serverSession.getSessionId()) } } } private class DeliveryReceiptTask implements Runnable { private final SMPPServerSession session private final SubmitSm submitSm private MessageId messageId public DeliveryReceiptTask(SMPPServerSession session, SubmitSm submitSm, MessageId messageId) { this.session = session this.submitSm = submitSm this.messageId = messageId } @Override public void run() { try { Thread.sleep(1000) } catch (InterruptedException e1) { e1.printStackTrace() } String stringValue = Integer.valueOf(messageId.getValue(), 16).toString() try { DeliveryReceipt delRec = new DeliveryReceipt(stringValue, 1, 1, new Date(), new Date(), DeliveryReceiptState.DELIVRD, null, new String(submitSm.getShortMessage())) session.deliverShortMessage( "mc", TypeOfNumber.valueOf(submitSm.getDestAddrTon()), NumberingPlanIndicator.valueOf(submitSm.getDestAddrNpi()), submitSm.getDestAddress(), TypeOfNumber.valueOf(submitSm.getSourceAddrTon()), NumberingPlanIndicator.valueOf(submitSm.getSourceAddrNpi()), submitSm.getSourceAddr(), new ESMClass(MessageMode.DEFAULT, MessageType.SMSC_DEL_RECEIPT, GSMSpecificFeature.DEFAULT), (byte) 0, (byte) 0, new RegisteredDelivery(0), DataCoding.newInstance(0), delRec.toString().getBytes()) logger.debug("Sending delivery reciept for message id " + messageId + ":" + stringValue) } catch (Exception e) { logger.error("Failed sending delivery_receipt for message id " + messageId + ":" + stringValue, e) } } } public static void main(String[] args) { int port try { port = Integer.parseInt(System.getProperty("jsmpp.simulator.port", DEFAULT_PORT.toString())) } catch (NumberFormatException e) { port = DEFAULT_PORT } BasicConfigurator.configure() SMPPServerSimulator smppServerSim = new SMPPServerSimulator(port) smppServerSim.run() } } Processor is running with no errors but the results of the script is not as expected, (port 8057 should become ready to bind, Code acts as SMS GW server) Any help please ? Note that the above script is working if I execute it as groovy code from Netbeans IDE ! ... View more