Spring Data Redis - PubSub: channel is not passed correctly to POJO message listener

[Loke]

Hi,

I am playing around with the pub-sub implementation of Spring Data Redis (version 1.0.1) and I cannot pass the channel correctly to my message listener. My problem is a bit similar than the one mentioned in this Jira issue: https://jira.springsource.org/browse/DATAREDIS-77.

The source code of my message listener is following:

public class ContactMessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(ContactMessageListener.class);

    public void handleMessage(Contact contact, String channel) {
        LOGGER.debug(channel + ":" + contact.toString());
    }
}

The configuration of my message listener adapter and the message listener container is following:

   @Bean
    public MessageListenerAdapter messageListenerAdapter() {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new ContactMessageListener());
        messageListenerAdapter.setSerializer(contactSerializer());
        return messageListenerAdapter;
    }

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer() {
        RedisMessageListenerContainer listeners = new RedisMessageListenerContainer();

        listeners.setConnectionFactory(redisConnectionFactory());
        listeners.addMessageListener(messageListenerAdapter(), Arrays.asList(
                new ChannelTopic(RedisContactService.CHANNEL_NEW_CONTACTS),
                new ChannelTopic(RedisContactService.CHANNEL_UPDATED_CONTACTS),
                new ChannelTopic(RedisContactService.CHANNEL_REMOVED_CONTACTS)
        ));

        return listeners;
    }

Then I send message to a channel that notifies about new contacts:

    @Override
    public Contact add(Contact added) {
        LOGGER.debug("Adding contact with information: " + added);

        //Persist contact etc...
        redisTemplate.convertAndSend(CHANNEL_NEW_CONTACTS, added);

        return added;
    }

However, in my log I get following message:

DEBUG - ContactMessageListener     - null:example.model.Contact@1eac9126[id=5,address=example.model.Address@536d6585[country=,streetAddress=,postCode=,postOffice=,state=],emailAddress=,firstName=Foo,lastName=Bar,phoneNumber=]

I ran this example on debugger and found out that the onMessage() method of MessageListenerAdapter class does not use channel information stored in a Message class. Instead it passes on the pattern that is always null. The source code of the onMessage() method is following (I added some questions in the comments by using CAPS):

public void onMessage(Message message, byte[] pattern) {
		try {
			// Check whether the delegate is a MessageListener impl itself.
			// In that case, the adapter will simply act as a pass-through.
                        //THIS WORKS 
			if (delegate != this) {
				if (delegate instanceof MessageListener) {
					((MessageListener) delegate).onMessage(message, pattern);
					return;
				}
			}

			// Regular case: find a handler method reflectively.
			Object convertedMessage = extractMessage(message);
                        //SHOULD THIS BE: String convertedChannel = stringSerializer.deSerialize(message.getChannel());?
			String convertedChannel = stringSerializer.deserialize(pattern);
			// Invoke the handler method with appropriate arguments.
			Object[] listenerArguments = new Object[] { convertedMessage, convertedChannel };

			invokeListenerMethod(invoker.getMethodName(), listenerArguments);
		} catch (Throwable th) {
			handleListenerException(th);
		}
	}

Am I missing something or should I just give up and implement the MessageListener interface?

[Jon Brisbin]

Looks like like line 724 in RedisMessageListenerContainer is the culprit. It's hard-coded to pass null as the pattern for some reason.

[Loke]

Looks like like line 724 in RedisMessageListenerContainer is the culprit. It's hard-coded to pass null as the pattern for some reason.

Hi Jon,

Thanks for your response. It is good to know that I did not mess up. However, the problem is that I still have to figure out a way to get the channel information. I would like to use the POJO message listener approach because I don't have to convert the message back to Contact object myself. Of course I could add channel information to the listener class as a constructor parameter but this sounds like a kludge to me. Any ideas?

[Jon Brisbin]

Looks like we need a JIRA issue. The following code in RedisMessageListenerContainer should be modified as follows.

processMessage(messageListener, message, null);

Should actually be:

processMessage(messageListener, message, message.getChannel());

[Jon Brisbin]

However, the problem is that I still have to figure out a way to get the channel information. I would like to use the POJO message listener approach because I don't have to convert the message back to Contact object myself. Of course I could add channel information to the listener class as a constructor parameter but this sounds like a kludge to me. Any ideas?

To be honest, I don't see it happening with the current code. If you're okay with using a snapshot, we could probably get that fix in right away. If not, it looks like you'll have to keep track of that yourself until the fix becomes available.

[Loke]

To be honest, I don't see it happening with the current code. If you're okay with using a snapshot, we could probably get that fix in right away. If not, it looks like you'll have to keep track of that yourself until the fix becomes available.

I understand this. I will create a Jira issue to the Spring Data Redis project and follow its progress myself.

Update: https://jira.springsource.org/browse/DATAREDIS-98

And thanks for the help Jon!