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 .