void processAck ( Message msg ) { PubAck pa ; Exception ex = null ; try { pa = PubAck . parseFrom ( msg . getData ( ) ) ; } catch ( InvalidProtocolBufferException e ) { // If we are speaking to a server we don't understand, let the // user know. System . err . println ( "Protocol error: " + e . getStackTrace ( ) ) ; return ; } // Remove AckClosure ackClosure = removeAck ( pa . getGuid ( ) ) ; if ( ackClosure != null ) { // Capture error if it exists. String ackError = pa . getError ( ) ; if ( ackClosure . ah != null ) { if ( ! ackError . isEmpty ( ) ) { ex = new IOException ( ackError ) ; } // Perform the ackHandler callback ackClosure . ah . onAck ( pa . getGuid ( ) , ex ) ; } else if ( ackClosure . ch != null ) { try { ackClosure . ch . put ( ackError ) ; } catch ( InterruptedException e ) { // ignore } } } }
Process an ack from the STAN cluster