Thanks Mark for pointing me to GetTwitter example. I will try to follow this pattern. PropertyDescriptor is only available in onTrigger method via ProcessContext. I needed to get hold of Service from initialize method using ProcessorInitializationContexts as documented in the Developer guide. this helps me to do expensive Service calls once not doing every time onTrigger method is invoked.
Thanks Sumo > On Sep 28, 2015, at 5:32 AM, Mark Payne <marka...@hotmail.com> wrote: > > Sumo, > > The preferred mechanism for obtaining a Controller Service is to do so via a > PropertyDescriptor. You would > specify that the property represents a Controller Service by using the > identifiesControllerService(Class) method. > This is discussed in the "Interacting with a Controller Service" section of > the Developer's Guide [1]. > > In terms of communicating with some asynchronous process, generally the best > solution is to use a queue and > then read from that queue in the onTrigger method. You will not be able to > call the onTrigger method properly > yourself, as it is designed to be called by the framework at the appropriate > time. I would recommend using > a BlockingQueue with a small capacity, as you do not want your Java Heap to > fill up with objects from this > Processor if the Processor is stopped for a while. You can look at how this > is done in the GetTwitter processor, > if you would like to have an example to look at. > > Let us know if you have any more questions, or if anything is still not clear! > > Thanks > -Mark > > > [1] > http://nifi.apache.org/docs/nifi-docs/html/developer-guide.html#interacting-with-controller-service > > > >> On Sep 27, 2015, at 3:53 PM, Sumanth Chinthagunta <xmlk...@gmail.com> wrote: >> >> >> Hi All, >> I am new to NiFi and I'm stuck with couple issues: >> >> 1. Unable to get hold of ControllerService from Processor’s init method. >> I wanted to pre-set some dependencies in my Processor during init phase >> instead of querying them repeatedly in onTrigger method. >> I am getting null for service and not sure what I have to pass for >> 'serviceIdentifier’ . I couldn't find documentation or examples on how to >> give Identifier to a service. >> >> >> final VertxServiceInterface vertxService = (VertxServiceInterface) >> context.getControllerServiceLookup().getControllerService("VertxService”) >> >> >> https://github.com/xmlking/nifi-websocket/blob/master/src/main/java/com/crossbusiness/nifi/processors/PutEventBus.java#L55 >> >> <https://github.com/xmlking/nifi-websocket/blob/master/src/main/java/com/crossbusiness/nifi/processors/PutEventBus.java#L55> >> >> 2. for my usecase I get data published to a topic from EventBus with >> following code. >> >> EventBus eb = vertx.eventBus(); >> >> eb.consumer("news.uk.sport", message -> { >> System.out.println("I have received a message: " + >> message.body()); >> }); >> >> I am working on a date ingest processor (push based) that needs to >> listen for new messages on a topic and send them to flow as FlowFile. >> In my case data source is EvenBus that expose emit messages via >> callback API. >> I am looking for ideas on how to call Processor’s onTrigger method when >> the above callback is evoked. >> Should I have to use my own intermediate queue and poll it in onTrigger >> method? >> is there a better way to trigger the onTrigger method programmatically >> ? >> >> Thanks >> Sumo >