This page provides sample code showing how to integrate the Grouper Messaging System with another messaging system.
- Interface GrouperMessagingSystem defines a messaging system. Implement this to integrate Grouper messaging with a messaging system
- GrouperBuiltinMessagingSystem implements GrouperMessagingSystem - Grouper's implementation of the messaging interface
- GrouperMessagingEngine is the Grouper API to send a message to various messaging systems
Configure messaging systems in the grouper.client.properties
################################ ## Grouper Messaging System ################################ # name of messaging system which is the default grouper.messaging.default.name.of.messaging.system = grouperBuiltinMessaging # name of a messaging system. note, "grouperBuiltinMessaging" can be arbitrary grouper.messaging.system.grouperBuiltinMessaging.name = grouperBuiltinMessaging # class that implements edu.internet2.middleware.grouperClient.messaging.GrouperMessagingSystem grouper.messaging.system.grouperBuiltinMessaging.class = edu.internet2.middleware.grouper.messaging.GrouperBuiltinMessagingSystem # name of a messaging system. note, "myAwsMessagingSystem" can be arbitrary # grouper.messaging.system.myAwsMessagingSystem.name = aws # class that implements edu.internet2.middleware.grouperClient.messaging.GrouperMessagingSystem # grouper.messaging.system.myAwsMessagingSystem.class =
Configure Builtin queues/topics
// create objects
GrouperBuiltinMessagingSystem.createQueue("abc");
GrouperBuiltinMessagingSystem.createTopic("def");
// permissions on objects
GrouperBuiltinMessagingSystem.allowSendToQueue("abc", SubjectTestHelper.SUBJ0);
GrouperBuiltinMessagingSystem.allowSendToTopic("abc", SubjectTestHelper.SUBJ0);
GrouperBuiltinMessagingSystem.allowReceiveFromQueue("abc", SubjectTestHelper.SUBJ0);
//link up topic with queue(s)
GrouperBuiltinMessagingSystem.topicSendsToQueue("def", "abc");
Use messaging
// send a message
GrouperMessageSendResult grouperMessageSendResult = GrouperMessagingEngine.send(new GrouperMessageSendParam().assignGrouperMessageSystemName("someName").assignQueueOrTopic("abc").addMessageBody("message body"));
// receive messages
GrouperMessageReceiveResult grouperMessageReceiveResult = GrouperMessagingEngine.receive(new GrouperMessageReceiveParam().assignGrouperMessageSystemName("someName").assignQueueOrTopic("abc"));
// mark as processed
GrouperMessageProcessedResult grouperMessageProcessedResult = GrouperMessagingEngine.markAsProcessed(new GrouperMessageProcessedParam().assignGrouperMessageSystemName("someName").assignQueueOrTopic("abc").addGrouperMessage(grouperMessage));
Messaging listener
##################################### ## Messaging listener ##################################### # note, change "messagingListener" in key to be the name of the listener. e.g. messaging.listener.myAzureListener.class # extends edu.internet2.middleware.grouper.messaging.MessagingListenerBase # this listener will just print out messages: edu.internet2.middleware.grouper.messaging.MessagingListenerPrint # #messaging.listener.messagingListener.class = edu.internet2.middleware.grouper.messaging.MessagingListener #messaging.listener.messagingListener.quartzCron = 0 * * * * ? #messaging.listener.messagingListener.messagingSystemName = grouperBuiltinMessaging #messaging.listener.messagingListener.queue = abc #messaging.listener.messagingListener.numberOfTriesPerIteration = 3 #messaging.listener.messagingListener.pollingTimeoutSeconds = 18 #messaging.listener.messagingListener.sleepSecondsInBetweenIterations = 0 #messaging.listener.messagingListener.maxMessagesToReceiveAtOnce = 20 # if there are 20 messages to receive at once, then do this 50 times per call max #messaging.listener.messagingListener.maxOuterLoops = 50
Sample messaging listener implementation
/**
* @author mchyzer
* $Id$
*/
package edu.internet2.middleware.grouper.messaging;
import java.util.Collection;
import edu.internet2.middleware.grouper.changeLog.ChangeLogEntry;
import edu.internet2.middleware.grouperClient.messaging.GrouperMessage;
import edu.internet2.middleware.grouperClient.messaging.GrouperMessageProcessedParam;
import edu.internet2.middleware.grouperClient.messaging.GrouperMessagingEngine;
/**
*
*/
public class MessagingListenerPrint extends MessagingListenerBase {
/**
*
*/
public MessagingListenerPrint() {
}
/**
* @see edu.internet2.middleware.grouper.messaging.MessagingListenerBase#processMessages(java.lang.String, java.lang.String, java.util.Collection, edu.internet2.middleware.grouper.messaging.MessagingListenerMetadata)
*/
@Override
public String processMessages(String messageSystemName, String queue,
Collection<GrouperMessage> grouperMessageList,
MessagingListenerMetadata messagingListenerMetadata) {
String lastIdProcessed = null;
for (GrouperMessage grouperMessage : grouperMessageList) {
try {
String json = grouperMessage.getMessageBody();
//try to convert to change log entry
try {
Collection<ChangeLogEntry> changeLogEntries = ChangeLogEntry.fromJsonToCollection(json);
for (ChangeLogEntry changeLogEntry : changeLogEntries) {
System.out.println("Change log entry: " + changeLogEntry.getChangeLogType().getChangeLogCategory() +
" -> " + changeLogEntry.getChangeLogType().getActionName() + ", " + changeLogEntry.getId());
}
System.out.println("Change log entry: " + json);
} catch (Exception e) {
System.out.println("Not change log entry: " + grouperMessage.getId() + ", " + json);
}
//mark it as processed
GrouperMessagingEngine.markAsProcessed(new GrouperMessageProcessedParam()
.assignQueue(queue).assignGropuerMessageSystemName(messageSystemName)
.addGrouperMessage(grouperMessage));
} catch (Exception e) {
messagingListenerMetadata.registerProblem(e, "Problem in message: " + grouperMessage.getId(), grouperMessage.getId());
break;
}
lastIdProcessed = grouperMessage.getId();
}
return lastIdProcessed;
}
}
Change log consumer that sends to messaging
##################################### ## Messaging integration with change log ##################################### # note, change "messaging" in key to be the name of the consumer. e.g. changeLog.consumer.myAzureConsumer.class #changeLog.consumer.messaging.class = edu.internet2.middleware.grouper.changeLog.ChangeLogConsumerToMessage #changeLog.consumer.messaging.quartzCron = 0 * * * * ? #changeLog.consumer.messaging.messagingSystemName = grouperBuiltinMessaging #changeLog.consumer.messaging.queueOrTopic = abc
Change log JSON
{
"event":[
{
"changeLogTypeId":"7db1fb2d34944668bc7d4485f1c7854b",
"contextId":"7e4ca9c1dfd14111a60c8b057e5ae5aa",
"createdOnDb":1458577616745723,
"sequenceNumber":473,
"changeLogTypeCategory":"privilege",
"changeLogTypeAction":"addPrivilege",
"field_id":"435e7190d1814d86a26e6bbd70e9bca3",
"field_privilegeName":"read",
"field_subjectId":"GrouperAll",
"field_sourceId":"g:isa",
"field_privilegeType":"access",
"field_ownerType":"group",
"field_ownerId":"7d09cfff2b444db696a0771b224081ef",
"field_ownerName":"test:testGroup3",
"field_memberId":"e6e154ea21f64c0d9ecb9f4137ab1ac3",
"field_fieldId":"5da0071e39ff4fb2a17cec73ac25efb3",
"field_membershipType":"flattened"
}
]
}
Convert Change Log Entry to and from json
String json = changeLogEntry.toJson(true); ChangeLogEntry newEntry = ChangeLogEntry.fromJsonToCollection(json).iterator().next();
Messaging ESB change log consumer, configure in grouper-loader.properties
##################################### ## Messaging integration ##################################### # note, change "messagingEsb" in key to be the name of the consumer. e.g. changeLog.consumer.myAzureConsumer.class #changeLog.consumer.messagingEsb.class = edu.internet2.middleware.grouper.changeLog.esb.consumer.EsbConsumer #changeLog.consumer.messagingEsb.quartzCron = 0 * * * * ? #changeLog.consumer.messagingEsb.elfilter = event.eventType eq 'GROUP_DELETE' || event.eventType eq 'GROUP_ADD' || event.eventType eq 'MEMBERSHIP_DELETE' || event.eventType eq 'MEMBERSHIP_ADD' #changeLog.consumer.messagingEsb.publisher.class = edu.internet2.middleware.grouper.changeLog.esb.consumer.EsbMessagingPublisher #changeLog.consumer.messagingEsb.publisher.messagingSystemName = grouperBuiltinMessaging #changeLog.consumer.messagingEsb.publisher.queueOrTopic = abc
Configure builtin messaging cleanup jobs in grouper-loader.properties
################################## ## grouper builtin messaging cleanup cron ################################## #quartz cron-like schedule for grouper messaging daemon. #leave blank to disable this, the default is every hour, 10 minutes after the hour #this daemon does cleanup on the builtin messaging table changeLog.builtinMessagingDaemon.quartz.cron = 0 10 * * * ? # after three days of not consuming messages, delete them, if -1, dont run this daemon grouper.builtin.messaging.deleteAllMessagesMoreThanHoursOld = 72 # after three hours of having processed messages, delete them. Note, if this is -1 just delete when marking processed grouper.builtin.messaging.deleteProcessedMessagesMoreThanMinutesOld = 180
Test messaging
Run the test:
Set this in log4j.properties to see debug and performance info:
log4j.logger.edu.internet2.middleware.grouper.messaging.GrouperBuiltinMessagingSystemTest = DEBUG
See Also