Unable to autowire beans when implementing pubsub using java client

[ANKITHA PRABHU]

Hi,

In the below code, I have implemented MessageReceiver and overrid the receiveMessage() method. While processing the message I need to call autowired classes, which are null when the message is read in receiveMessage() method. Is there any solution for this ? 

  

Thanks,

Ankitha Prabhu

@Component

public class ActivityReceiver<T> implements MessageReceiver{

@Autowired

BeanFinder beanFinder;

@Override

public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {

       System.out.println("Id: " + message.getMessageId());

       CloudEventImpl<T> ce = CloudEventMapper.mapToGenericCloudEvent(message);

        System.out.println("Data: " + ce.getAttributes().getType() +" "+ ce.getData());

        IAbstractEventProcessor<PubsubMessage> eventProcessor =  beanFinder.get(CloudEventMapper.getEventType(ce).get().getEventName());

eventProcessor.execute(message);

        consumer.ack();

}

}

[Kir Titievsky]

Ankitha, It seems that   "beanFinder.get(CloudEventMapper.getEventType(ce).get().getEventName())" part of your code returns "null". Correct?  If, so does this work any other context, outside the message receiver?

--
You received this message because you are subscribed to the Google Groups "Google Cloud Pub/Sub Discussions" group.
To unsubscribe from this group and stop receiving emails from it, send an email to cloud-pubsub-dis...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/cloud-pubsub-discuss/63df3bf2-f667-4965-a42c-f49e630c27d6n%40googlegroups.com.

--

Kir Titievsky | Product Manager | Google Cloud Pub/Sub 

[ANKITHA PRABHU]

Yes it works for other context outside the message receiver in the same class. 

Thanks,

Ankitha Prabhu 

[Kir Titievsky]

It looks like neither 'ce' nor 'beanFinder' are initialized in this scope.  I don't know the real answer, but I suspect you'll have to think about the lifecycle of these objects. Since message handler is called asynchronously either might have expired or worse been concurrently altered by another  part of the code. 

[ANKITHA PRABHU]

Hey, I found my error. 

I was using ApplicationRunner to start my subscriber as follows: 

@Component

public class ApplicationStartupRunner implements ApplicationRunner {

@Override

public void run(ApplicationArguments args) throws Exception {

Subscriber subscriber = Subscriber

.newBuilder(ProjectSubscriptionName.of("festive-utility-297015", "app-activity"), new ActivityReceiver()).build();

subscriber.startAsync().awaitRunning();

System.out.printf("Listening for messages on %s:\n",

ProjectSubscriptionName.of("festive-utility-297015", "app-activity").toString());

subscriber.awaitTerminated();

}

} 

Because if the new ActivityReceiver() which implements MessageReceiver class, the underlying autowired classes were initialized to null. What I should have done is as follows and fixed my issue: 

@Component

public class ApplicationStartupRunner implements ApplicationRunner {

@Autowired 

ActivityReceiver activityReceiver;

@Override

public void run(ApplicationArguments args) throws Exception {

Subscriber subscriber = Subscriber

.newBuilder(ProjectSubscriptionName.of("festive-utility-297015", "app-activity"), activityReceiver).build();

subscriber.startAsync().awaitRunning();

System.out.printf("Listening for messages on %s:\n",

ProjectSubscriptionName.of("festive-utility-297015", "app-activity").toString());

subscriber.awaitTerminated();

}

} 

Thanks a lot for your time. 

Ankitha Prabhu