public void handleMessage(Message<?> message) {
        if (logger.isDebugEnabled()) {
            logger.debug("received Integration Message: " + message);
        }
        AsyncMessage flexMessage = new AsyncMessage();
        flexMessage.setBody(message.getPayload());
        MessageHeaders headers = message.getHeaders();
        flexMessage.setMessageId(headers.containsKey(FlexHeaders.MESSAGE_ID) ? headers.get(FlexHeaders.MESSAGE_ID, String.class) : headers.getId().toString());
        Long timestamp = headers.containsKey(FlexHeaders.TIMESTAMP) ? Long.parseLong(headers.get(FlexHeaders.TIMESTAMP, String.class)) : headers.getTimestamp();
        flexMessage.setTimestamp(timestamp);
        Long expirationDate = headers.getExpirationDate();
        if (expirationDate != null) {
            flexMessage.setTimeToLive(expirationDate - timestamp);
        }
        if (headers.containsKey(FlexHeaders.MESSAGE_CLIENT_ID)) {
        	flexMessage.setClientId(headers.get(FlexHeaders.MESSAGE_CLIENT_ID));
        }
        for (Map.Entry<String, Object> header : headers.entrySet()) {
            String key = header.getKey();
            if (!filteredHeaders.contains(key)) {
                flexMessage.setHeader(key, header.getValue());
            }
        }
        flexMessage.setDestination(this.getDestination().getId());
        MessageService messageService = (MessageService) getDestination().getService();
        if (filterSender && headers.containsKey(FlexHeaders.FLEX_CLIENT_ID)) {
        	Set<Object> subscribers = new HashSet<Object>(this.subscriberIds);
        	FlexClient flexClient = messageService.getMessageBroker().getFlexClientManager().getFlexClient(headers.get(FlexHeaders.FLEX_CLIENT_ID).toString());
        	for (Object subscriberId : this.subscriberIds) {
        		if (flexClient.getMessageClient(subscriberId.toString()) != null) {
        			subscribers.remove(subscriberId);
        		}
        	}
        	messageService.pushMessageToClients(subscribers, flexMessage, true);
        } else {
        	messageService.pushMessageToClients(flexMessage, true);
        }
        messageService.sendPushMessageFromPeer(flexMessage, true);
    }
public void handleMessage ( Message < ? > message ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "received Integration Message: " + message ) ; } AsyncMessage flexMessage = new AsyncMessage ( ) ; flexMessage . setBody ( message . getPayload ( ) ) ; MessageHeaders headers = message . getHeaders ( ) ; flexMessage . setMessageId ( headers . containsKey ( FlexHeaders . MESSAGE_ID ) ? headers . get ( FlexHeaders . MESSAGE_ID , String . class ) : headers . getId ( ) . toString ( ) ) ; Long timestamp = headers . containsKey ( FlexHeaders . TIMESTAMP ) ? Long . parseLong ( headers . get ( FlexHeaders . TIMESTAMP , String . class ) ) : headers . getTimestamp ( ) ; flexMessage . setTimestamp ( timestamp ) ; Long expirationDate = headers . getExpirationDate ( ) ; if ( expirationDate != null ) { flexMessage . setTimeToLive ( expirationDate - timestamp ) ; } if ( headers . containsKey ( FlexHeaders . MESSAGE_CLIENT_ID ) ) { flexMessage . setClientId ( headers . get ( FlexHeaders . MESSAGE_CLIENT_ID ) ) ; } for ( Map . Entry < String , Object > header : headers . entrySet ( ) ) { String key = header . getKey ( ) ; if ( ! filteredHeaders . contains ( key ) ) { flexMessage . setHeader ( key , header . getValue ( ) ) ; } } flexMessage . setDestination ( this . getDestination ( ) . getId ( ) ) ; MessageService messageService = ( MessageService ) getDestination ( ) . getService ( ) ; if ( filterSender && headers . containsKey ( FlexHeaders . FLEX_CLIENT_ID ) ) { Set < Object > subscribers = new HashSet < Object > ( this . subscriberIds ) ; FlexClient flexClient = messageService . getMessageBroker ( ) . getFlexClientManager ( ) . getFlexClient ( headers . get ( FlexHeaders . FLEX_CLIENT_ID ) . toString ( ) ) ; for ( Object subscriberId : this . subscriberIds ) { if ( flexClient . getMessageClient ( subscriberId . toString ( ) ) != null ) { subscribers . remove ( subscriberId ) ; } } messageService . pushMessageToClients ( subscribers , flexMessage , true ) ; } else { messageService . pushMessageToClients ( flexMessage , true ) ; } messageService . sendPushMessageFromPeer ( flexMessage , true ) ; }
Invoked when a Message is received from the Spring Integration channel.