001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.transport.stomp; 018 019import java.io.BufferedReader; 020import java.io.IOException; 021import java.io.InputStream; 022import java.io.InputStreamReader; 023import java.io.OutputStreamWriter; 024import java.io.PrintWriter; 025import java.util.HashMap; 026import java.util.Iterator; 027import java.util.Map; 028import java.util.concurrent.ConcurrentHashMap; 029import java.util.concurrent.ConcurrentMap; 030import java.util.concurrent.atomic.AtomicBoolean; 031 032import javax.jms.JMSException; 033 034import org.apache.activemq.ActiveMQPrefetchPolicy; 035import org.apache.activemq.advisory.AdvisorySupport; 036import org.apache.activemq.broker.BrokerContext; 037import org.apache.activemq.broker.BrokerContextAware; 038import org.apache.activemq.command.ActiveMQDestination; 039import org.apache.activemq.command.ActiveMQMessage; 040import org.apache.activemq.command.ActiveMQTempQueue; 041import org.apache.activemq.command.ActiveMQTempTopic; 042import org.apache.activemq.command.Command; 043import org.apache.activemq.command.CommandTypes; 044import org.apache.activemq.command.ConnectionError; 045import org.apache.activemq.command.ConnectionId; 046import org.apache.activemq.command.ConnectionInfo; 047import org.apache.activemq.command.ConsumerControl; 048import org.apache.activemq.command.ConsumerId; 049import org.apache.activemq.command.ConsumerInfo; 050import org.apache.activemq.command.DestinationInfo; 051import org.apache.activemq.command.ExceptionResponse; 052import org.apache.activemq.command.LocalTransactionId; 053import org.apache.activemq.command.MessageAck; 054import org.apache.activemq.command.MessageDispatch; 055import org.apache.activemq.command.MessageId; 056import org.apache.activemq.command.ProducerId; 057import org.apache.activemq.command.ProducerInfo; 058import org.apache.activemq.command.RemoveSubscriptionInfo; 059import org.apache.activemq.command.Response; 060import org.apache.activemq.command.SessionId; 061import org.apache.activemq.command.SessionInfo; 062import org.apache.activemq.command.ShutdownInfo; 063import org.apache.activemq.command.TransactionId; 064import org.apache.activemq.command.TransactionInfo; 065import org.apache.activemq.util.ByteArrayOutputStream; 066import org.apache.activemq.util.FactoryFinder; 067import org.apache.activemq.util.IOExceptionSupport; 068import org.apache.activemq.util.IdGenerator; 069import org.apache.activemq.util.IntrospectionSupport; 070import org.apache.activemq.util.LongSequenceGenerator; 071import org.slf4j.Logger; 072import org.slf4j.LoggerFactory; 073 074/** 075 * @author <a href="http://hiramchirino.com">chirino</a> 076 */ 077public class ProtocolConverter { 078 079 private static final Logger LOG = LoggerFactory.getLogger(ProtocolConverter.class); 080 081 private static final IdGenerator CONNECTION_ID_GENERATOR = new IdGenerator(); 082 083 private static final String BROKER_VERSION; 084 private static final StompFrame ping = new StompFrame(Stomp.Commands.KEEPALIVE); 085 086 static { 087 String version = "5.6.0"; 088 try(InputStream in = ProtocolConverter.class.getResourceAsStream("/org/apache/activemq/version.txt")) { 089 if (in != null) { 090 try(InputStreamReader isr = new InputStreamReader(in); 091 BufferedReader reader = new BufferedReader(isr)) { 092 version = reader.readLine(); 093 } 094 } 095 }catch(Exception e) { 096 } 097 098 BROKER_VERSION = version; 099 } 100 101 private final ConnectionId connectionId = new ConnectionId(CONNECTION_ID_GENERATOR.generateId()); 102 private final SessionId sessionId = new SessionId(connectionId, -1); 103 private final ProducerId producerId = new ProducerId(sessionId, 1); 104 105 private final LongSequenceGenerator consumerIdGenerator = new LongSequenceGenerator(); 106 private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator(); 107 private final LongSequenceGenerator transactionIdGenerator = new LongSequenceGenerator(); 108 private final LongSequenceGenerator tempDestinationGenerator = new LongSequenceGenerator(); 109 110 private final ConcurrentMap<Integer, ResponseHandler> resposeHandlers = new ConcurrentHashMap<Integer, ResponseHandler>(); 111 private final ConcurrentMap<ConsumerId, StompSubscription> subscriptionsByConsumerId = new ConcurrentHashMap<ConsumerId, StompSubscription>(); 112 private final ConcurrentMap<String, StompSubscription> subscriptions = new ConcurrentHashMap<String, StompSubscription>(); 113 private final ConcurrentMap<String, ActiveMQDestination> tempDestinations = new ConcurrentHashMap<String, ActiveMQDestination>(); 114 private final ConcurrentMap<String, String> tempDestinationAmqToStompMap = new ConcurrentHashMap<String, String>(); 115 private final Map<String, LocalTransactionId> transactions = new ConcurrentHashMap<String, LocalTransactionId>(); 116 private final StompTransport stompTransport; 117 118 private final ConcurrentMap<String, AckEntry> pedingAcks = new ConcurrentHashMap<String, AckEntry>(); 119 private final IdGenerator ACK_ID_GENERATOR = new IdGenerator(); 120 121 private final Object commnadIdMutex = new Object(); 122 private int lastCommandId; 123 private final AtomicBoolean connected = new AtomicBoolean(false); 124 private final FrameTranslator frameTranslator = new LegacyFrameTranslator(); 125 private final FactoryFinder FRAME_TRANSLATOR_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/frametranslator/"); 126 private final BrokerContext brokerContext; 127 private String version = "1.0"; 128 private long hbReadInterval; 129 private long hbWriteInterval; 130 private float hbGracePeriodMultiplier = 1.0f; 131 private String defaultHeartBeat = Stomp.DEFAULT_HEART_BEAT; 132 133 private static class AckEntry { 134 135 private final String messageId; 136 private final StompSubscription subscription; 137 138 public AckEntry(String messageId, StompSubscription subscription) { 139 this.messageId = messageId; 140 this.subscription = subscription; 141 } 142 143 public MessageAck onMessageAck(TransactionId transactionId) { 144 return subscription.onStompMessageAck(messageId, transactionId); 145 } 146 147 public MessageAck onMessageNack(TransactionId transactionId) throws ProtocolException { 148 return subscription.onStompMessageNack(messageId, transactionId); 149 } 150 151 public String getMessageId() { 152 return this.messageId; 153 } 154 155 @SuppressWarnings("unused") 156 public StompSubscription getSubscription() { 157 return this.subscription; 158 } 159 } 160 161 public ProtocolConverter(StompTransport stompTransport, BrokerContext brokerContext) { 162 this.stompTransport = stompTransport; 163 this.brokerContext = brokerContext; 164 } 165 166 protected int generateCommandId() { 167 synchronized (commnadIdMutex) { 168 return lastCommandId++; 169 } 170 } 171 172 protected ResponseHandler createResponseHandler(final StompFrame command) { 173 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 174 if (receiptId != null) { 175 return new ResponseHandler() { 176 @Override 177 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 178 if (response.isException()) { 179 // Generally a command can fail.. but that does not invalidate the connection. 180 // We report back the failure but we don't close the connection. 181 Throwable exception = ((ExceptionResponse)response).getException(); 182 handleException(exception, command); 183 } else { 184 StompFrame sc = new StompFrame(); 185 sc.setAction(Stomp.Responses.RECEIPT); 186 sc.setHeaders(new HashMap<String, String>(1)); 187 sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 188 stompTransport.sendToStomp(sc); 189 } 190 } 191 }; 192 } 193 return null; 194 } 195 196 protected void sendToActiveMQ(Command command, ResponseHandler handler) { 197 command.setCommandId(generateCommandId()); 198 if (handler != null) { 199 command.setResponseRequired(true); 200 resposeHandlers.put(Integer.valueOf(command.getCommandId()), handler); 201 } 202 stompTransport.sendToActiveMQ(command); 203 } 204 205 protected void sendToStomp(StompFrame command) throws IOException { 206 stompTransport.sendToStomp(command); 207 } 208 209 protected FrameTranslator findTranslator(String header) { 210 return findTranslator(header, null, false); 211 } 212 213 protected FrameTranslator findTranslator(String header, ActiveMQDestination destination, boolean advisory) { 214 FrameTranslator translator = frameTranslator; 215 try { 216 if (header != null) { 217 translator = (FrameTranslator) FRAME_TRANSLATOR_FINDER.newInstance(header); 218 } else { 219 if (destination != null && (advisory || AdvisorySupport.isAdvisoryTopic(destination))) { 220 translator = new JmsFrameTranslator(); 221 } 222 } 223 } catch (Exception ignore) { 224 // if anything goes wrong use the default translator 225 } 226 227 if (translator instanceof BrokerContextAware) { 228 ((BrokerContextAware)translator).setBrokerContext(brokerContext); 229 } 230 231 return translator; 232 } 233 234 /** 235 * Convert a STOMP command 236 * 237 * @param command 238 */ 239 public void onStompCommand(StompFrame command) throws IOException, JMSException { 240 try { 241 242 if (command.getClass() == StompFrameError.class) { 243 throw ((StompFrameError)command).getException(); 244 } 245 246 String action = command.getAction(); 247 if (action.startsWith(Stomp.Commands.SEND)) { 248 onStompSend(command); 249 } else if (action.startsWith(Stomp.Commands.ACK)) { 250 onStompAck(command); 251 } else if (action.startsWith(Stomp.Commands.NACK)) { 252 onStompNack(command); 253 } else if (action.startsWith(Stomp.Commands.BEGIN)) { 254 onStompBegin(command); 255 } else if (action.startsWith(Stomp.Commands.COMMIT)) { 256 onStompCommit(command); 257 } else if (action.startsWith(Stomp.Commands.ABORT)) { 258 onStompAbort(command); 259 } else if (action.startsWith(Stomp.Commands.SUBSCRIBE)) { 260 onStompSubscribe(command); 261 } else if (action.startsWith(Stomp.Commands.UNSUBSCRIBE)) { 262 onStompUnsubscribe(command); 263 } else if (action.startsWith(Stomp.Commands.CONNECT) || 264 action.startsWith(Stomp.Commands.STOMP)) { 265 onStompConnect(command); 266 } else if (action.startsWith(Stomp.Commands.DISCONNECT)) { 267 onStompDisconnect(command); 268 } else { 269 throw new ProtocolException("Unknown STOMP action: " + action); 270 } 271 272 } catch (ProtocolException e) { 273 handleException(e, command); 274 // Some protocol errors can cause the connection to get closed. 275 if (e.isFatal()) { 276 getStompTransport().onException(e); 277 } 278 } 279 } 280 281 protected void handleException(Throwable exception, StompFrame command) throws IOException { 282 LOG.warn("Exception occurred processing: \n" + command + ": " + exception.toString()); 283 if (LOG.isDebugEnabled()) { 284 LOG.debug("Exception detail", exception); 285 } 286 287 // Let the stomp client know about any protocol errors. 288 ByteArrayOutputStream baos = new ByteArrayOutputStream(); 289 PrintWriter stream = new PrintWriter(new OutputStreamWriter(baos, "UTF-8")); 290 exception.printStackTrace(stream); 291 stream.close(); 292 293 HashMap<String, String> headers = new HashMap<String, String>(); 294 headers.put(Stomp.Headers.Error.MESSAGE, exception.getMessage()); 295 headers.put(Stomp.Headers.CONTENT_TYPE, "text/plain"); 296 297 if (command != null) { 298 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 299 if (receiptId != null) { 300 headers.put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 301 } 302 } 303 304 StompFrame errorMessage = new StompFrame(Stomp.Responses.ERROR, headers, baos.toByteArray()); 305 sendToStomp(errorMessage); 306 } 307 308 protected void onStompSend(StompFrame command) throws IOException, JMSException { 309 checkConnected(); 310 311 Map<String, String> headers = command.getHeaders(); 312 String destination = headers.get(Stomp.Headers.Send.DESTINATION); 313 if (destination == null) { 314 throw new ProtocolException("SEND received without a Destination specified!"); 315 } 316 317 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 318 headers.remove("transaction"); 319 320 ActiveMQMessage message = convertMessage(command); 321 322 message.setProducerId(producerId); 323 MessageId id = new MessageId(producerId, messageIdGenerator.getNextSequenceId()); 324 message.setMessageId(id); 325 326 if (stompTx != null) { 327 TransactionId activemqTx = transactions.get(stompTx); 328 if (activemqTx == null) { 329 throw new ProtocolException("Invalid transaction id: " + stompTx); 330 } 331 message.setTransactionId(activemqTx); 332 } 333 334 message.onSend(); 335 sendToActiveMQ(message, createResponseHandler(command)); 336 } 337 338 protected void onStompNack(StompFrame command) throws ProtocolException { 339 340 checkConnected(); 341 342 if (this.version.equals(Stomp.V1_0)) { 343 throw new ProtocolException("NACK received but connection is in v1.0 mode."); 344 } 345 346 Map<String, String> headers = command.getHeaders(); 347 348 String subscriptionId = headers.get(Stomp.Headers.Ack.SUBSCRIPTION); 349 if (subscriptionId == null && !this.version.equals(Stomp.V1_2)) { 350 throw new ProtocolException("NACK received without a subscription id for acknowledge!"); 351 } 352 353 String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID); 354 if (messageId == null && !this.version.equals(Stomp.V1_2)) { 355 throw new ProtocolException("NACK received without a message-id to acknowledge!"); 356 } 357 358 String ackId = headers.get(Stomp.Headers.Ack.ACK_ID); 359 if (ackId == null && this.version.equals(Stomp.V1_2)) { 360 throw new ProtocolException("NACK received without an ack header to acknowledge!"); 361 } 362 363 TransactionId activemqTx = null; 364 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 365 if (stompTx != null) { 366 activemqTx = transactions.get(stompTx); 367 if (activemqTx == null) { 368 throw new ProtocolException("Invalid transaction id: " + stompTx); 369 } 370 } 371 372 boolean nacked = false; 373 374 if (ackId != null) { 375 AckEntry pendingAck = this.pedingAcks.remove(ackId); 376 if (pendingAck != null) { 377 messageId = pendingAck.getMessageId(); 378 MessageAck ack = pendingAck.onMessageNack(activemqTx); 379 if (ack != null) { 380 sendToActiveMQ(ack, createResponseHandler(command)); 381 nacked = true; 382 } 383 } 384 } else if (subscriptionId != null) { 385 StompSubscription sub = this.subscriptions.get(subscriptionId); 386 if (sub != null) { 387 MessageAck ack = sub.onStompMessageNack(messageId, activemqTx); 388 if (ack != null) { 389 sendToActiveMQ(ack, createResponseHandler(command)); 390 nacked = true; 391 } 392 } 393 } 394 395 if (!nacked) { 396 throw new ProtocolException("Unexpected NACK received for message-id [" + messageId + "]"); 397 } 398 } 399 400 protected void onStompAck(StompFrame command) throws ProtocolException { 401 checkConnected(); 402 403 Map<String, String> headers = command.getHeaders(); 404 String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID); 405 if (messageId == null && !(this.version.equals(Stomp.V1_2))) { 406 throw new ProtocolException("ACK received without a message-id to acknowledge!"); 407 } 408 409 String subscriptionId = headers.get(Stomp.Headers.Ack.SUBSCRIPTION); 410 if (subscriptionId == null && this.version.equals(Stomp.V1_1)) { 411 throw new ProtocolException("ACK received without a subscription id for acknowledge!"); 412 } 413 414 String ackId = headers.get(Stomp.Headers.Ack.ACK_ID); 415 if (ackId == null && this.version.equals(Stomp.V1_2)) { 416 throw new ProtocolException("ACK received without a ack id for acknowledge!"); 417 } 418 419 TransactionId activemqTx = null; 420 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 421 if (stompTx != null) { 422 activemqTx = transactions.get(stompTx); 423 if (activemqTx == null) { 424 throw new ProtocolException("Invalid transaction id: " + stompTx); 425 } 426 } 427 428 boolean acked = false; 429 430 if (ackId != null) { 431 AckEntry pendingAck = this.pedingAcks.remove(ackId); 432 if (pendingAck != null) { 433 messageId = pendingAck.getMessageId(); 434 MessageAck ack = pendingAck.onMessageAck(activemqTx); 435 if (ack != null) { 436 sendToActiveMQ(ack, createResponseHandler(command)); 437 acked = true; 438 } 439 } 440 441 } else if (subscriptionId != null) { 442 StompSubscription sub = this.subscriptions.get(subscriptionId); 443 if (sub != null) { 444 MessageAck ack = sub.onStompMessageAck(messageId, activemqTx); 445 if (ack != null) { 446 sendToActiveMQ(ack, createResponseHandler(command)); 447 acked = true; 448 } 449 } 450 } else { 451 // STOMP v1.0: acking with just a message id is very bogus since the same message id 452 // could have been sent to 2 different subscriptions on the same Stomp connection. 453 // For example, when 2 subs are created on the same topic. 454 for (StompSubscription sub : subscriptionsByConsumerId.values()) { 455 MessageAck ack = sub.onStompMessageAck(messageId, activemqTx); 456 if (ack != null) { 457 sendToActiveMQ(ack, createResponseHandler(command)); 458 acked = true; 459 break; 460 } 461 } 462 } 463 464 if (!acked) { 465 throw new ProtocolException("Unexpected ACK received for message-id [" + messageId + "]"); 466 } 467 } 468 469 protected void onStompBegin(StompFrame command) throws ProtocolException { 470 checkConnected(); 471 472 Map<String, String> headers = command.getHeaders(); 473 474 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 475 476 if (!headers.containsKey(Stomp.Headers.TRANSACTION)) { 477 throw new ProtocolException("Must specify the transaction you are beginning"); 478 } 479 480 if (transactions.get(stompTx) != null) { 481 throw new ProtocolException("The transaction was already started: " + stompTx); 482 } 483 484 LocalTransactionId activemqTx = new LocalTransactionId(connectionId, transactionIdGenerator.getNextSequenceId()); 485 transactions.put(stompTx, activemqTx); 486 487 TransactionInfo tx = new TransactionInfo(); 488 tx.setConnectionId(connectionId); 489 tx.setTransactionId(activemqTx); 490 tx.setType(TransactionInfo.BEGIN); 491 492 sendToActiveMQ(tx, createResponseHandler(command)); 493 } 494 495 protected void onStompCommit(StompFrame command) throws ProtocolException { 496 checkConnected(); 497 498 Map<String, String> headers = command.getHeaders(); 499 500 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 501 if (stompTx == null) { 502 throw new ProtocolException("Must specify the transaction you are committing"); 503 } 504 505 TransactionId activemqTx = transactions.remove(stompTx); 506 if (activemqTx == null) { 507 throw new ProtocolException("Invalid transaction id: " + stompTx); 508 } 509 510 for (StompSubscription sub : subscriptionsByConsumerId.values()) { 511 sub.onStompCommit(activemqTx); 512 } 513 514 pedingAcks.clear(); 515 516 TransactionInfo tx = new TransactionInfo(); 517 tx.setConnectionId(connectionId); 518 tx.setTransactionId(activemqTx); 519 tx.setType(TransactionInfo.COMMIT_ONE_PHASE); 520 521 sendToActiveMQ(tx, createResponseHandler(command)); 522 } 523 524 protected void onStompAbort(StompFrame command) throws ProtocolException { 525 checkConnected(); 526 Map<String, String> headers = command.getHeaders(); 527 528 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 529 if (stompTx == null) { 530 throw new ProtocolException("Must specify the transaction you are committing"); 531 } 532 533 TransactionId activemqTx = transactions.remove(stompTx); 534 if (activemqTx == null) { 535 throw new ProtocolException("Invalid transaction id: " + stompTx); 536 } 537 for (StompSubscription sub : subscriptionsByConsumerId.values()) { 538 try { 539 sub.onStompAbort(activemqTx); 540 } catch (Exception e) { 541 throw new ProtocolException("Transaction abort failed", false, e); 542 } 543 } 544 545 pedingAcks.clear(); 546 547 TransactionInfo tx = new TransactionInfo(); 548 tx.setConnectionId(connectionId); 549 tx.setTransactionId(activemqTx); 550 tx.setType(TransactionInfo.ROLLBACK); 551 552 sendToActiveMQ(tx, createResponseHandler(command)); 553 } 554 555 protected void onStompSubscribe(StompFrame command) throws ProtocolException { 556 checkConnected(); 557 FrameTranslator translator = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)); 558 Map<String, String> headers = command.getHeaders(); 559 560 String subscriptionId = headers.get(Stomp.Headers.Subscribe.ID); 561 String destination = headers.get(Stomp.Headers.Subscribe.DESTINATION); 562 563 if (!this.version.equals(Stomp.V1_0) && subscriptionId == null) { 564 throw new ProtocolException("SUBSCRIBE received without a subscription id!"); 565 } 566 567 final ActiveMQDestination actualDest = translator.convertDestination(this, destination, true); 568 569 if (actualDest == null) { 570 throw new ProtocolException("Invalid 'null' Destination."); 571 } 572 573 final ConsumerId id = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId()); 574 ConsumerInfo consumerInfo = new ConsumerInfo(id); 575 consumerInfo.setPrefetchSize(actualDest.isQueue() ? 576 ActiveMQPrefetchPolicy.DEFAULT_QUEUE_PREFETCH : 577 headers.containsKey("activemq.subscriptionName") ? 578 ActiveMQPrefetchPolicy.DEFAULT_DURABLE_TOPIC_PREFETCH : ActiveMQPrefetchPolicy.DEFAULT_TOPIC_PREFETCH); 579 consumerInfo.setDispatchAsync(true); 580 581 String browser = headers.get(Stomp.Headers.Subscribe.BROWSER); 582 if (browser != null && browser.equals(Stomp.TRUE)) { 583 584 if (this.version.equals(Stomp.V1_0)) { 585 throw new ProtocolException("Queue Browser feature only valid for Stomp v1.1+ clients!"); 586 } 587 588 consumerInfo.setBrowser(true); 589 consumerInfo.setPrefetchSize(ActiveMQPrefetchPolicy.DEFAULT_QUEUE_BROWSER_PREFETCH); 590 } 591 592 String selector = headers.remove(Stomp.Headers.Subscribe.SELECTOR); 593 if (selector != null) { 594 consumerInfo.setSelector("convert_string_expressions:" + selector); 595 } 596 597 IntrospectionSupport.setProperties(consumerInfo, headers, "activemq."); 598 599 if (actualDest.isQueue() && consumerInfo.getSubscriptionName() != null) { 600 throw new ProtocolException("Invalid Subscription: cannot durably subscribe to a Queue destination!"); 601 } 602 603 consumerInfo.setDestination(actualDest); 604 605 StompSubscription stompSubscription; 606 if (!consumerInfo.isBrowser()) { 607 stompSubscription = new StompSubscription(this, subscriptionId, consumerInfo, headers.get(Stomp.Headers.TRANSFORMATION)); 608 } else { 609 stompSubscription = new StompQueueBrowserSubscription(this, subscriptionId, consumerInfo, headers.get(Stomp.Headers.TRANSFORMATION)); 610 } 611 stompSubscription.setDestination(actualDest); 612 613 String ackMode = headers.get(Stomp.Headers.Subscribe.ACK_MODE); 614 if (Stomp.Headers.Subscribe.AckModeValues.CLIENT.equals(ackMode)) { 615 stompSubscription.setAckMode(StompSubscription.CLIENT_ACK); 616 } else if (Stomp.Headers.Subscribe.AckModeValues.INDIVIDUAL.equals(ackMode)) { 617 stompSubscription.setAckMode(StompSubscription.INDIVIDUAL_ACK); 618 } else { 619 stompSubscription.setAckMode(StompSubscription.AUTO_ACK); 620 } 621 622 subscriptionsByConsumerId.put(id, stompSubscription); 623 // Stomp v1.0 doesn't need to set this header so we avoid an NPE if not set. 624 if (subscriptionId != null) { 625 subscriptions.put(subscriptionId, stompSubscription); 626 } 627 628 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 629 if (receiptId != null && consumerInfo.getPrefetchSize() > 0) { 630 631 final StompFrame cmd = command; 632 final int prefetch = consumerInfo.getPrefetchSize(); 633 634 // Since dispatch could beat the receipt we set prefetch to zero to start and then 635 // once we've sent our Receipt we are safe to turn on dispatch if the response isn't 636 // an error message. 637 consumerInfo.setPrefetchSize(0); 638 639 final ResponseHandler handler = new ResponseHandler() { 640 @Override 641 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 642 if (response.isException()) { 643 // Generally a command can fail.. but that does not invalidate the connection. 644 // We report back the failure but we don't close the connection. 645 Throwable exception = ((ExceptionResponse)response).getException(); 646 handleException(exception, cmd); 647 } else { 648 StompFrame sc = new StompFrame(); 649 sc.setAction(Stomp.Responses.RECEIPT); 650 sc.setHeaders(new HashMap<String, String>(1)); 651 sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 652 stompTransport.sendToStomp(sc); 653 654 ConsumerControl control = new ConsumerControl(); 655 control.setPrefetch(prefetch); 656 control.setDestination(actualDest); 657 control.setConsumerId(id); 658 659 sendToActiveMQ(control, null); 660 } 661 } 662 }; 663 664 sendToActiveMQ(consumerInfo, handler); 665 } else { 666 sendToActiveMQ(consumerInfo, createResponseHandler(command)); 667 } 668 } 669 670 protected void onStompUnsubscribe(StompFrame command) throws ProtocolException { 671 checkConnected(); 672 Map<String, String> headers = command.getHeaders(); 673 674 ActiveMQDestination destination = null; 675 Object o = headers.get(Stomp.Headers.Unsubscribe.DESTINATION); 676 if (o != null) { 677 destination = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)).convertDestination(this, (String)o, true); 678 } 679 680 String subscriptionId = headers.get(Stomp.Headers.Unsubscribe.ID); 681 if (!this.version.equals(Stomp.V1_0) && subscriptionId == null) { 682 throw new ProtocolException("UNSUBSCRIBE received without a subscription id!"); 683 } 684 685 if (subscriptionId == null && destination == null) { 686 throw new ProtocolException("Must specify the subscriptionId or the destination you are unsubscribing from"); 687 } 688 689 // check if it is a durable subscription 690 String durable = command.getHeaders().get("activemq.subscriptionName"); 691 String clientId = durable; 692 if (!this.version.equals(Stomp.V1_0)) { 693 clientId = connectionInfo.getClientId(); 694 } 695 696 if (durable != null) { 697 RemoveSubscriptionInfo info = new RemoveSubscriptionInfo(); 698 info.setClientId(clientId); 699 info.setSubscriptionName(durable); 700 info.setConnectionId(connectionId); 701 sendToActiveMQ(info, createResponseHandler(command)); 702 return; 703 } 704 705 if (subscriptionId != null) { 706 707 StompSubscription sub = this.subscriptions.remove(subscriptionId); 708 if (sub != null) { 709 sendToActiveMQ(sub.getConsumerInfo().createRemoveCommand(), createResponseHandler(command)); 710 return; 711 } 712 713 } else { 714 715 // Unsubscribing using a destination is a bit weird if multiple subscriptions 716 // are created with the same destination. 717 for (Iterator<StompSubscription> iter = subscriptionsByConsumerId.values().iterator(); iter.hasNext();) { 718 StompSubscription sub = iter.next(); 719 if (destination != null && destination.equals(sub.getDestination())) { 720 sendToActiveMQ(sub.getConsumerInfo().createRemoveCommand(), createResponseHandler(command)); 721 iter.remove(); 722 return; 723 } 724 } 725 } 726 727 throw new ProtocolException("No subscription matched."); 728 } 729 730 ConnectionInfo connectionInfo = new ConnectionInfo(); 731 732 protected void onStompConnect(final StompFrame command) throws ProtocolException { 733 734 if (connected.get()) { 735 throw new ProtocolException("Already connected."); 736 } 737 738 final Map<String, String> headers = command.getHeaders(); 739 740 // allow anyone to login for now 741 String login = headers.get(Stomp.Headers.Connect.LOGIN); 742 String passcode = headers.get(Stomp.Headers.Connect.PASSCODE); 743 String clientId = headers.get(Stomp.Headers.Connect.CLIENT_ID); 744 String heartBeat = headers.get(Stomp.Headers.Connect.HEART_BEAT); 745 746 if (heartBeat == null) { 747 heartBeat = defaultHeartBeat; 748 } 749 750 this.version = StompCodec.detectVersion(headers); 751 752 configureInactivityMonitor(heartBeat.trim()); 753 754 IntrospectionSupport.setProperties(connectionInfo, headers, "activemq."); 755 connectionInfo.setConnectionId(connectionId); 756 if (clientId != null) { 757 connectionInfo.setClientId(clientId); 758 } else { 759 connectionInfo.setClientId("" + connectionInfo.getConnectionId().toString()); 760 } 761 762 connectionInfo.setResponseRequired(true); 763 connectionInfo.setUserName(login); 764 connectionInfo.setPassword(passcode); 765 connectionInfo.setTransportContext(command.getTransportContext()); 766 767 sendToActiveMQ(connectionInfo, new ResponseHandler() { 768 @Override 769 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 770 771 if (response.isException()) { 772 // If the connection attempt fails we close the socket. 773 Throwable exception = ((ExceptionResponse)response).getException(); 774 handleException(exception, command); 775 getStompTransport().onException(IOExceptionSupport.create(exception)); 776 return; 777 } 778 779 final SessionInfo sessionInfo = new SessionInfo(sessionId); 780 sendToActiveMQ(sessionInfo, null); 781 782 final ProducerInfo producerInfo = new ProducerInfo(producerId); 783 sendToActiveMQ(producerInfo, new ResponseHandler() { 784 @Override 785 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 786 787 if (response.isException()) { 788 // If the connection attempt fails we close the socket. 789 Throwable exception = ((ExceptionResponse)response).getException(); 790 handleException(exception, command); 791 getStompTransport().onException(IOExceptionSupport.create(exception)); 792 } 793 794 connected.set(true); 795 HashMap<String, String> responseHeaders = new HashMap<String, String>(); 796 797 responseHeaders.put(Stomp.Headers.Connected.SESSION, connectionInfo.getClientId()); 798 String requestId = headers.get(Stomp.Headers.Connect.REQUEST_ID); 799 if (requestId == null) { 800 // TODO legacy 801 requestId = headers.get(Stomp.Headers.RECEIPT_REQUESTED); 802 } 803 if (requestId != null) { 804 // TODO legacy 805 responseHeaders.put(Stomp.Headers.Connected.RESPONSE_ID, requestId); 806 responseHeaders.put(Stomp.Headers.Response.RECEIPT_ID, requestId); 807 } 808 809 responseHeaders.put(Stomp.Headers.Connected.VERSION, version); 810 responseHeaders.put(Stomp.Headers.Connected.HEART_BEAT, 811 String.format("%d,%d", hbWriteInterval, hbReadInterval)); 812 responseHeaders.put(Stomp.Headers.Connected.SERVER, "ActiveMQ/"+BROKER_VERSION); 813 814 StompFrame sc = new StompFrame(); 815 sc.setAction(Stomp.Responses.CONNECTED); 816 sc.setHeaders(responseHeaders); 817 sendToStomp(sc); 818 819 StompWireFormat format = stompTransport.getWireFormat(); 820 if (format != null) { 821 format.setStompVersion(version); 822 } 823 } 824 }); 825 } 826 }); 827 } 828 829 protected void onStompDisconnect(StompFrame command) throws ProtocolException { 830 if (connected.get()) { 831 sendToActiveMQ(connectionInfo.createRemoveCommand(), createResponseHandler(command)); 832 sendToActiveMQ(new ShutdownInfo(), createResponseHandler(command)); 833 connected.set(false); 834 } 835 } 836 837 protected void checkConnected() throws ProtocolException { 838 if (!connected.get()) { 839 throw new ProtocolException("Not connected."); 840 } 841 } 842 843 /** 844 * Dispatch a ActiveMQ command 845 * 846 * @param command 847 * @throws IOException 848 */ 849 public void onActiveMQCommand(Command command) throws IOException, JMSException { 850 if (command.isResponse()) { 851 Response response = (Response)command; 852 ResponseHandler rh = resposeHandlers.remove(Integer.valueOf(response.getCorrelationId())); 853 if (rh != null) { 854 rh.onResponse(this, response); 855 } else { 856 // Pass down any unexpected errors. Should this close the connection? 857 if (response.isException()) { 858 Throwable exception = ((ExceptionResponse)response).getException(); 859 handleException(exception, null); 860 } 861 } 862 } else if (command.isMessageDispatch()) { 863 MessageDispatch md = (MessageDispatch)command; 864 StompSubscription sub = subscriptionsByConsumerId.get(md.getConsumerId()); 865 if (sub != null) { 866 String ackId = null; 867 if (version.equals(Stomp.V1_2) && sub.getAckMode() != Stomp.Headers.Subscribe.AckModeValues.AUTO && md.getMessage() != null) { 868 AckEntry pendingAck = new AckEntry(md.getMessage().getMessageId().toString(), sub); 869 ackId = this.ACK_ID_GENERATOR.generateId(); 870 this.pedingAcks.put(ackId, pendingAck); 871 } 872 try { 873 sub.onMessageDispatch(md, ackId); 874 } catch (Exception ex) { 875 if (ackId != null) { 876 this.pedingAcks.remove(ackId); 877 } 878 } 879 } 880 } else if (command.getDataStructureType() == CommandTypes.KEEP_ALIVE_INFO) { 881 stompTransport.sendToStomp(ping); 882 } else if (command.getDataStructureType() == ConnectionError.DATA_STRUCTURE_TYPE) { 883 // Pass down any unexpected async errors. Should this close the connection? 884 Throwable exception = ((ConnectionError)command).getException(); 885 handleException(exception, null); 886 } 887 } 888 889 public ActiveMQMessage convertMessage(StompFrame command) throws IOException, JMSException { 890 ActiveMQMessage msg = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)).convertFrame(this, command); 891 return msg; 892 } 893 894 public StompFrame convertMessage(ActiveMQMessage message, boolean ignoreTransformation) throws IOException, JMSException { 895 if (ignoreTransformation == true) { 896 return frameTranslator.convertMessage(this, message); 897 } else { 898 FrameTranslator translator = findTranslator( 899 message.getStringProperty(Stomp.Headers.TRANSFORMATION), message.getDestination(), message.isAdvisory()); 900 return translator.convertMessage(this, message); 901 } 902 } 903 904 public StompTransport getStompTransport() { 905 return stompTransport; 906 } 907 908 public ActiveMQDestination createTempDestination(String name, boolean topic) { 909 ActiveMQDestination rc = tempDestinations.get(name); 910 if( rc == null ) { 911 if (topic) { 912 rc = new ActiveMQTempTopic(connectionId, tempDestinationGenerator.getNextSequenceId()); 913 } else { 914 rc = new ActiveMQTempQueue(connectionId, tempDestinationGenerator.getNextSequenceId()); 915 } 916 sendToActiveMQ(new DestinationInfo(connectionId, DestinationInfo.ADD_OPERATION_TYPE, rc), null); 917 tempDestinations.put(name, rc); 918 tempDestinationAmqToStompMap.put(rc.getQualifiedName(), name); 919 } 920 return rc; 921 } 922 923 public String getCreatedTempDestinationName(ActiveMQDestination destination) { 924 return tempDestinationAmqToStompMap.get(destination.getQualifiedName()); 925 } 926 927 public String getDefaultHeartBeat() { 928 return defaultHeartBeat; 929 } 930 931 public void setDefaultHeartBeat(String defaultHeartBeat) { 932 this.defaultHeartBeat = defaultHeartBeat; 933 } 934 935 /** 936 * @return the hbGracePeriodMultiplier 937 */ 938 public float getHbGracePeriodMultiplier() { 939 return hbGracePeriodMultiplier; 940 } 941 942 /** 943 * @param hbGracePeriodMultiplier the hbGracePeriodMultiplier to set 944 */ 945 public void setHbGracePeriodMultiplier(float hbGracePeriodMultiplier) { 946 this.hbGracePeriodMultiplier = hbGracePeriodMultiplier; 947 } 948 949 protected void configureInactivityMonitor(String heartBeatConfig) throws ProtocolException { 950 951 String[] keepAliveOpts = heartBeatConfig.split(Stomp.COMMA); 952 953 if (keepAliveOpts == null || keepAliveOpts.length != 2) { 954 throw new ProtocolException("Invalid heart-beat header:" + heartBeatConfig, true); 955 } else { 956 957 try { 958 hbReadInterval = (Long.parseLong(keepAliveOpts[0])); 959 hbWriteInterval = Long.parseLong(keepAliveOpts[1]); 960 } catch(NumberFormatException e) { 961 throw new ProtocolException("Invalid heart-beat header:" + heartBeatConfig, true); 962 } 963 964 try { 965 StompInactivityMonitor monitor = this.stompTransport.getInactivityMonitor(); 966 monitor.setReadCheckTime((long) (hbReadInterval * hbGracePeriodMultiplier)); 967 monitor.setInitialDelayTime(Math.min(hbReadInterval, hbWriteInterval)); 968 monitor.setWriteCheckTime(hbWriteInterval); 969 monitor.startMonitoring(); 970 } catch(Exception ex) { 971 hbReadInterval = 0; 972 hbWriteInterval = 0; 973 } 974 975 if (LOG.isDebugEnabled()) { 976 LOG.debug("Stomp Connect heartbeat conf RW[" + hbReadInterval + "," + hbWriteInterval + "]"); 977 } 978 } 979 } 980 981 protected void sendReceipt(StompFrame command) { 982 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 983 if (receiptId != null) { 984 StompFrame sc = new StompFrame(); 985 sc.setAction(Stomp.Responses.RECEIPT); 986 sc.setHeaders(new HashMap<String, String>(1)); 987 sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 988 try { 989 sendToStomp(sc); 990 } catch (IOException e) { 991 LOG.warn("Could not send a receipt for " + command, e); 992 } 993 } 994 } 995}