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, true); 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 if (command == null) { 283 LOG.warn("Exception occurred while processing a command: {}", exception.toString()); 284 } else { 285 LOG.warn("Exception occurred processing: {} -> {}", safeGetAction(command), exception.toString()); 286 } 287 288 if (LOG.isDebugEnabled()) { 289 LOG.debug("Exception detail", exception); 290 } 291 292 if (command != null && LOG.isTraceEnabled()) { 293 LOG.trace("Command that caused the error: {}", command); 294 } 295 296 // Let the stomp client know about any protocol errors. 297 ByteArrayOutputStream baos = new ByteArrayOutputStream(); 298 PrintWriter stream = new PrintWriter(new OutputStreamWriter(baos, "UTF-8")); 299 exception.printStackTrace(stream); 300 stream.close(); 301 302 HashMap<String, String> headers = new HashMap<String, String>(); 303 headers.put(Stomp.Headers.Error.MESSAGE, exception.getMessage()); 304 headers.put(Stomp.Headers.CONTENT_TYPE, "text/plain"); 305 306 if (command != null) { 307 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 308 if (receiptId != null) { 309 headers.put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 310 } 311 } 312 313 StompFrame errorMessage = new StompFrame(Stomp.Responses.ERROR, headers, baos.toByteArray()); 314 sendToStomp(errorMessage); 315 } 316 317 protected void onStompSend(StompFrame command) throws IOException, JMSException { 318 checkConnected(); 319 320 Map<String, String> headers = command.getHeaders(); 321 String destination = headers.get(Stomp.Headers.Send.DESTINATION); 322 if (destination == null) { 323 throw new ProtocolException("SEND received without a Destination specified!"); 324 } 325 326 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 327 headers.remove("transaction"); 328 329 ActiveMQMessage message = convertMessage(command); 330 331 message.setProducerId(producerId); 332 MessageId id = new MessageId(producerId, messageIdGenerator.getNextSequenceId()); 333 message.setMessageId(id); 334 335 if (stompTx != null) { 336 TransactionId activemqTx = transactions.get(stompTx); 337 if (activemqTx == null) { 338 throw new ProtocolException("Invalid transaction id: " + stompTx); 339 } 340 message.setTransactionId(activemqTx); 341 } 342 343 message.onSend(); 344 message.beforeMarshall(null); 345 sendToActiveMQ(message, createResponseHandler(command)); 346 } 347 348 protected void onStompNack(StompFrame command) throws ProtocolException { 349 350 checkConnected(); 351 352 if (this.version.equals(Stomp.V1_0)) { 353 throw new ProtocolException("NACK received but connection is in v1.0 mode."); 354 } 355 356 Map<String, String> headers = command.getHeaders(); 357 358 String subscriptionId = headers.get(Stomp.Headers.Ack.SUBSCRIPTION); 359 if (subscriptionId == null && !this.version.equals(Stomp.V1_2)) { 360 throw new ProtocolException("NACK received without a subscription id for acknowledge!"); 361 } 362 363 String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID); 364 if (messageId == null && !this.version.equals(Stomp.V1_2)) { 365 throw new ProtocolException("NACK received without a message-id to acknowledge!"); 366 } 367 368 String ackId = headers.get(Stomp.Headers.Ack.ACK_ID); 369 if (ackId == null && this.version.equals(Stomp.V1_2)) { 370 throw new ProtocolException("NACK received without an ack header to acknowledge!"); 371 } 372 373 TransactionId activemqTx = null; 374 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 375 if (stompTx != null) { 376 activemqTx = transactions.get(stompTx); 377 if (activemqTx == null) { 378 throw new ProtocolException("Invalid transaction id: " + stompTx); 379 } 380 } 381 382 boolean nacked = false; 383 384 if (ackId != null) { 385 AckEntry pendingAck = this.pedingAcks.remove(ackId); 386 if (pendingAck != null) { 387 messageId = pendingAck.getMessageId(); 388 MessageAck ack = pendingAck.onMessageNack(activemqTx); 389 if (ack != null) { 390 sendToActiveMQ(ack, createResponseHandler(command)); 391 nacked = true; 392 } 393 } 394 } else if (subscriptionId != null) { 395 StompSubscription sub = this.subscriptions.get(subscriptionId); 396 if (sub != null) { 397 MessageAck ack = sub.onStompMessageNack(messageId, activemqTx); 398 if (ack != null) { 399 sendToActiveMQ(ack, createResponseHandler(command)); 400 nacked = true; 401 } 402 } 403 } 404 405 if (!nacked) { 406 throw new ProtocolException("Unexpected NACK received for message-id [" + messageId + "]"); 407 } 408 } 409 410 protected void onStompAck(StompFrame command) throws ProtocolException { 411 checkConnected(); 412 413 Map<String, String> headers = command.getHeaders(); 414 String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID); 415 if (messageId == null && !(this.version.equals(Stomp.V1_2))) { 416 throw new ProtocolException("ACK received without a message-id to acknowledge!"); 417 } 418 419 String subscriptionId = headers.get(Stomp.Headers.Ack.SUBSCRIPTION); 420 if (subscriptionId == null && this.version.equals(Stomp.V1_1)) { 421 throw new ProtocolException("ACK received without a subscription id for acknowledge!"); 422 } 423 424 String ackId = headers.get(Stomp.Headers.Ack.ACK_ID); 425 if (ackId == null && this.version.equals(Stomp.V1_2)) { 426 throw new ProtocolException("ACK received without a ack id for acknowledge!"); 427 } 428 429 TransactionId activemqTx = null; 430 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 431 if (stompTx != null) { 432 activemqTx = transactions.get(stompTx); 433 if (activemqTx == null) { 434 throw new ProtocolException("Invalid transaction id: " + stompTx); 435 } 436 } 437 438 boolean acked = false; 439 440 if (ackId != null) { 441 AckEntry pendingAck = this.pedingAcks.remove(ackId); 442 if (pendingAck != null) { 443 messageId = pendingAck.getMessageId(); 444 MessageAck ack = pendingAck.onMessageAck(activemqTx); 445 if (ack != null) { 446 sendToActiveMQ(ack, createResponseHandler(command)); 447 acked = true; 448 } 449 } 450 451 } else if (subscriptionId != null) { 452 StompSubscription sub = this.subscriptions.get(subscriptionId); 453 if (sub != null) { 454 MessageAck ack = sub.onStompMessageAck(messageId, activemqTx); 455 if (ack != null) { 456 sendToActiveMQ(ack, createResponseHandler(command)); 457 acked = true; 458 } 459 } 460 } else { 461 // STOMP v1.0: acking with just a message id is very bogus since the same message id 462 // could have been sent to 2 different subscriptions on the same Stomp connection. 463 // For example, when 2 subs are created on the same topic. 464 for (StompSubscription sub : subscriptionsByConsumerId.values()) { 465 MessageAck ack = sub.onStompMessageAck(messageId, activemqTx); 466 if (ack != null) { 467 sendToActiveMQ(ack, createResponseHandler(command)); 468 acked = true; 469 break; 470 } 471 } 472 } 473 474 if (!acked) { 475 throw new ProtocolException("Unexpected ACK received for message-id [" + messageId + "]"); 476 } 477 } 478 479 protected void onStompBegin(StompFrame command) throws ProtocolException { 480 checkConnected(); 481 482 Map<String, String> headers = command.getHeaders(); 483 484 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 485 486 if (!headers.containsKey(Stomp.Headers.TRANSACTION)) { 487 throw new ProtocolException("Must specify the transaction you are beginning"); 488 } 489 490 if (transactions.get(stompTx) != null) { 491 throw new ProtocolException("The transaction was already started: " + stompTx); 492 } 493 494 LocalTransactionId activemqTx = new LocalTransactionId(connectionId, transactionIdGenerator.getNextSequenceId()); 495 transactions.put(stompTx, activemqTx); 496 497 TransactionInfo tx = new TransactionInfo(); 498 tx.setConnectionId(connectionId); 499 tx.setTransactionId(activemqTx); 500 tx.setType(TransactionInfo.BEGIN); 501 502 sendToActiveMQ(tx, createResponseHandler(command)); 503 } 504 505 protected void onStompCommit(StompFrame command) throws ProtocolException { 506 checkConnected(); 507 508 Map<String, String> headers = command.getHeaders(); 509 510 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 511 if (stompTx == null) { 512 throw new ProtocolException("Must specify the transaction you are committing"); 513 } 514 515 TransactionId activemqTx = transactions.remove(stompTx); 516 if (activemqTx == null) { 517 throw new ProtocolException("Invalid transaction id: " + stompTx); 518 } 519 520 for (StompSubscription sub : subscriptionsByConsumerId.values()) { 521 sub.onStompCommit(activemqTx); 522 } 523 524 pedingAcks.clear(); 525 526 TransactionInfo tx = new TransactionInfo(); 527 tx.setConnectionId(connectionId); 528 tx.setTransactionId(activemqTx); 529 tx.setType(TransactionInfo.COMMIT_ONE_PHASE); 530 531 sendToActiveMQ(tx, createResponseHandler(command)); 532 } 533 534 protected void onStompAbort(StompFrame command) throws ProtocolException { 535 checkConnected(); 536 Map<String, String> headers = command.getHeaders(); 537 538 String stompTx = headers.get(Stomp.Headers.TRANSACTION); 539 if (stompTx == null) { 540 throw new ProtocolException("Must specify the transaction you are committing"); 541 } 542 543 TransactionId activemqTx = transactions.remove(stompTx); 544 if (activemqTx == null) { 545 throw new ProtocolException("Invalid transaction id: " + stompTx); 546 } 547 for (StompSubscription sub : subscriptionsByConsumerId.values()) { 548 try { 549 sub.onStompAbort(activemqTx); 550 } catch (Exception e) { 551 throw new ProtocolException("Transaction abort failed", false, e); 552 } 553 } 554 555 pedingAcks.clear(); 556 557 TransactionInfo tx = new TransactionInfo(); 558 tx.setConnectionId(connectionId); 559 tx.setTransactionId(activemqTx); 560 tx.setType(TransactionInfo.ROLLBACK); 561 562 sendToActiveMQ(tx, createResponseHandler(command)); 563 } 564 565 protected void onStompSubscribe(StompFrame command) throws ProtocolException { 566 checkConnected(); 567 FrameTranslator translator = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)); 568 Map<String, String> headers = command.getHeaders(); 569 570 String subscriptionId = headers.get(Stomp.Headers.Subscribe.ID); 571 String destination = headers.get(Stomp.Headers.Subscribe.DESTINATION); 572 573 if (!this.version.equals(Stomp.V1_0) && subscriptionId == null) { 574 throw new ProtocolException("SUBSCRIBE received without a subscription id!"); 575 } 576 577 final ActiveMQDestination actualDest = translator.convertDestination(this, destination, true); 578 579 if (actualDest == null) { 580 throw new ProtocolException("Invalid 'null' Destination."); 581 } 582 583 final ConsumerId id = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId()); 584 ConsumerInfo consumerInfo = new ConsumerInfo(id); 585 consumerInfo.setPrefetchSize(actualDest.isQueue() ? 586 ActiveMQPrefetchPolicy.DEFAULT_QUEUE_PREFETCH : 587 headers.containsKey("activemq.subscriptionName") ? 588 ActiveMQPrefetchPolicy.DEFAULT_DURABLE_TOPIC_PREFETCH : ActiveMQPrefetchPolicy.DEFAULT_TOPIC_PREFETCH); 589 consumerInfo.setDispatchAsync(true); 590 591 String browser = headers.get(Stomp.Headers.Subscribe.BROWSER); 592 if (browser != null && browser.equals(Stomp.TRUE)) { 593 594 if (this.version.equals(Stomp.V1_0)) { 595 throw new ProtocolException("Queue Browser feature only valid for Stomp v1.1+ clients!"); 596 } 597 598 consumerInfo.setBrowser(true); 599 consumerInfo.setPrefetchSize(ActiveMQPrefetchPolicy.DEFAULT_QUEUE_BROWSER_PREFETCH); 600 } 601 602 String selector = headers.remove(Stomp.Headers.Subscribe.SELECTOR); 603 if (selector != null) { 604 consumerInfo.setSelector("convert_string_expressions:" + selector); 605 } 606 607 IntrospectionSupport.setProperties(consumerInfo, headers, "activemq."); 608 609 if (actualDest.isQueue() && consumerInfo.getSubscriptionName() != null) { 610 throw new ProtocolException("Invalid Subscription: cannot durably subscribe to a Queue destination!"); 611 } 612 613 consumerInfo.setDestination(actualDest); 614 615 StompSubscription stompSubscription; 616 if (!consumerInfo.isBrowser()) { 617 stompSubscription = new StompSubscription(this, subscriptionId, consumerInfo, headers.get(Stomp.Headers.TRANSFORMATION)); 618 } else { 619 stompSubscription = new StompQueueBrowserSubscription(this, subscriptionId, consumerInfo, headers.get(Stomp.Headers.TRANSFORMATION)); 620 } 621 stompSubscription.setDestination(actualDest); 622 623 String ackMode = headers.get(Stomp.Headers.Subscribe.ACK_MODE); 624 if (Stomp.Headers.Subscribe.AckModeValues.CLIENT.equals(ackMode)) { 625 stompSubscription.setAckMode(StompSubscription.CLIENT_ACK); 626 } else if (Stomp.Headers.Subscribe.AckModeValues.INDIVIDUAL.equals(ackMode)) { 627 stompSubscription.setAckMode(StompSubscription.INDIVIDUAL_ACK); 628 } else { 629 stompSubscription.setAckMode(StompSubscription.AUTO_ACK); 630 } 631 632 subscriptionsByConsumerId.put(id, stompSubscription); 633 // Stomp v1.0 doesn't need to set this header so we avoid an NPE if not set. 634 if (subscriptionId != null) { 635 subscriptions.put(subscriptionId, stompSubscription); 636 } 637 638 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 639 if (receiptId != null && consumerInfo.getPrefetchSize() > 0) { 640 641 final StompFrame cmd = command; 642 final int prefetch = consumerInfo.getPrefetchSize(); 643 644 // Since dispatch could beat the receipt we set prefetch to zero to start and then 645 // once we've sent our Receipt we are safe to turn on dispatch if the response isn't 646 // an error message. 647 consumerInfo.setPrefetchSize(0); 648 649 final ResponseHandler handler = new ResponseHandler() { 650 @Override 651 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 652 if (response.isException()) { 653 // Generally a command can fail.. but that does not invalidate the connection. 654 // We report back the failure but we don't close the connection. 655 Throwable exception = ((ExceptionResponse)response).getException(); 656 handleException(exception, cmd); 657 } else { 658 StompFrame sc = new StompFrame(); 659 sc.setAction(Stomp.Responses.RECEIPT); 660 sc.setHeaders(new HashMap<String, String>(1)); 661 sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 662 stompTransport.sendToStomp(sc); 663 664 ConsumerControl control = new ConsumerControl(); 665 control.setPrefetch(prefetch); 666 control.setDestination(actualDest); 667 control.setConsumerId(id); 668 669 sendToActiveMQ(control, null); 670 } 671 } 672 }; 673 674 sendToActiveMQ(consumerInfo, handler); 675 } else { 676 sendToActiveMQ(consumerInfo, createResponseHandler(command)); 677 } 678 } 679 680 protected void onStompUnsubscribe(StompFrame command) throws ProtocolException { 681 checkConnected(); 682 Map<String, String> headers = command.getHeaders(); 683 684 ActiveMQDestination destination = null; 685 Object o = headers.get(Stomp.Headers.Unsubscribe.DESTINATION); 686 if (o != null) { 687 destination = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)).convertDestination(this, (String)o, true); 688 } 689 690 String subscriptionId = headers.get(Stomp.Headers.Unsubscribe.ID); 691 if (!this.version.equals(Stomp.V1_0) && subscriptionId == null) { 692 throw new ProtocolException("UNSUBSCRIBE received without a subscription id!"); 693 } 694 695 if (subscriptionId == null && destination == null) { 696 throw new ProtocolException("Must specify the subscriptionId or the destination you are unsubscribing from"); 697 } 698 699 // check if it is a durable subscription 700 String durable = command.getHeaders().get("activemq.subscriptionName"); 701 String clientId = durable; 702 if (!this.version.equals(Stomp.V1_0)) { 703 clientId = connectionInfo.getClientId(); 704 } 705 706 if (durable != null) { 707 RemoveSubscriptionInfo info = new RemoveSubscriptionInfo(); 708 info.setClientId(clientId); 709 info.setSubscriptionName(durable); 710 info.setConnectionId(connectionId); 711 sendToActiveMQ(info, createResponseHandler(command)); 712 return; 713 } 714 715 if (subscriptionId != null) { 716 StompSubscription sub = this.subscriptions.remove(subscriptionId); 717 if (sub != null) { 718 sendToActiveMQ(sub.getConsumerInfo().createRemoveCommand(), createResponseHandler(command)); 719 return; 720 } 721 } else { 722 // Unsubscribing using a destination is a bit weird if multiple subscriptions 723 // are created with the same destination. 724 for (Iterator<StompSubscription> iter = subscriptionsByConsumerId.values().iterator(); iter.hasNext();) { 725 StompSubscription sub = iter.next(); 726 if (destination != null && destination.equals(sub.getDestination())) { 727 sendToActiveMQ(sub.getConsumerInfo().createRemoveCommand(), createResponseHandler(command)); 728 iter.remove(); 729 return; 730 } 731 } 732 } 733 734 throw new ProtocolException("No subscription matched."); 735 } 736 737 ConnectionInfo connectionInfo = new ConnectionInfo(); 738 739 protected void onStompConnect(final StompFrame command) throws ProtocolException { 740 741 if (connected.get()) { 742 throw new ProtocolException("Already connected."); 743 } 744 745 final Map<String, String> headers = command.getHeaders(); 746 747 // allow anyone to login for now 748 String login = headers.get(Stomp.Headers.Connect.LOGIN); 749 String passcode = headers.get(Stomp.Headers.Connect.PASSCODE); 750 String clientId = headers.get(Stomp.Headers.Connect.CLIENT_ID); 751 String heartBeat = headers.get(Stomp.Headers.Connect.HEART_BEAT); 752 753 if (heartBeat == null) { 754 heartBeat = defaultHeartBeat; 755 } 756 757 this.version = StompCodec.detectVersion(headers); 758 759 configureInactivityMonitor(heartBeat.trim()); 760 761 IntrospectionSupport.setProperties(connectionInfo, headers, "activemq."); 762 connectionInfo.setConnectionId(connectionId); 763 if (clientId != null) { 764 connectionInfo.setClientId(clientId); 765 } else { 766 connectionInfo.setClientId("" + connectionInfo.getConnectionId().toString()); 767 } 768 769 connectionInfo.setResponseRequired(true); 770 connectionInfo.setUserName(login); 771 connectionInfo.setPassword(passcode); 772 connectionInfo.setTransportContext(command.getTransportContext()); 773 774 sendToActiveMQ(connectionInfo, new ResponseHandler() { 775 @Override 776 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 777 778 if (response.isException()) { 779 // If the connection attempt fails we close the socket. 780 Throwable exception = ((ExceptionResponse)response).getException(); 781 handleException(exception, command); 782 getStompTransport().onException(IOExceptionSupport.create(exception)); 783 return; 784 } 785 786 final SessionInfo sessionInfo = new SessionInfo(sessionId); 787 sendToActiveMQ(sessionInfo, null); 788 789 final ProducerInfo producerInfo = new ProducerInfo(producerId); 790 sendToActiveMQ(producerInfo, new ResponseHandler() { 791 @Override 792 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 793 794 if (response.isException()) { 795 // If the connection attempt fails we close the socket. 796 Throwable exception = ((ExceptionResponse)response).getException(); 797 handleException(exception, command); 798 getStompTransport().onException(IOExceptionSupport.create(exception)); 799 } 800 801 connected.set(true); 802 HashMap<String, String> responseHeaders = new HashMap<String, String>(); 803 804 responseHeaders.put(Stomp.Headers.Connected.SESSION, connectionInfo.getClientId()); 805 String requestId = headers.get(Stomp.Headers.Connect.REQUEST_ID); 806 if (requestId == null) { 807 // TODO legacy 808 requestId = headers.get(Stomp.Headers.RECEIPT_REQUESTED); 809 } 810 if (requestId != null) { 811 // TODO legacy 812 responseHeaders.put(Stomp.Headers.Connected.RESPONSE_ID, requestId); 813 responseHeaders.put(Stomp.Headers.Response.RECEIPT_ID, requestId); 814 } 815 816 responseHeaders.put(Stomp.Headers.Connected.VERSION, version); 817 responseHeaders.put(Stomp.Headers.Connected.HEART_BEAT, 818 String.format("%d,%d", hbWriteInterval, hbReadInterval)); 819 responseHeaders.put(Stomp.Headers.Connected.SERVER, "ActiveMQ/"+BROKER_VERSION); 820 821 StompFrame sc = new StompFrame(); 822 sc.setAction(Stomp.Responses.CONNECTED); 823 sc.setHeaders(responseHeaders); 824 sendToStomp(sc); 825 826 StompWireFormat format = stompTransport.getWireFormat(); 827 if (format != null) { 828 format.setStompVersion(version); 829 } 830 } 831 }); 832 } 833 }); 834 } 835 836 protected void onStompDisconnect(StompFrame command) throws ProtocolException { 837 if (connected.get()) { 838 sendToActiveMQ(connectionInfo.createRemoveCommand(), createResponseHandler(command)); 839 sendToActiveMQ(new ShutdownInfo(), createResponseHandler(command)); 840 connected.set(false); 841 } 842 } 843 844 protected void checkConnected() throws ProtocolException { 845 if (!connected.get()) { 846 throw new ProtocolException("Not connected."); 847 } 848 } 849 850 /** 851 * Dispatch a ActiveMQ command 852 * 853 * @param command 854 * @throws IOException 855 */ 856 public void onActiveMQCommand(Command command) throws IOException, JMSException { 857 if (command.isResponse()) { 858 Response response = (Response)command; 859 ResponseHandler rh = resposeHandlers.remove(Integer.valueOf(response.getCorrelationId())); 860 if (rh != null) { 861 rh.onResponse(this, response); 862 } else { 863 // Pass down any unexpected errors. Should this close the connection? 864 if (response.isException()) { 865 Throwable exception = ((ExceptionResponse)response).getException(); 866 handleException(exception, null); 867 } 868 } 869 } else if (command.isMessageDispatch()) { 870 MessageDispatch md = (MessageDispatch)command; 871 StompSubscription sub = subscriptionsByConsumerId.get(md.getConsumerId()); 872 if (sub != null) { 873 String ackId = null; 874 if (version.equals(Stomp.V1_2) && sub.getAckMode() != Stomp.Headers.Subscribe.AckModeValues.AUTO && md.getMessage() != null) { 875 AckEntry pendingAck = new AckEntry(md.getMessage().getMessageId().toString(), sub); 876 ackId = this.ACK_ID_GENERATOR.generateId(); 877 this.pedingAcks.put(ackId, pendingAck); 878 } 879 try { 880 sub.onMessageDispatch(md, ackId); 881 } catch (Exception ex) { 882 if (ackId != null) { 883 this.pedingAcks.remove(ackId); 884 } 885 } 886 } 887 } else if (command.getDataStructureType() == CommandTypes.KEEP_ALIVE_INFO) { 888 stompTransport.sendToStomp(ping); 889 } else if (command.getDataStructureType() == ConnectionError.DATA_STRUCTURE_TYPE) { 890 // Pass down any unexpected async errors. Should this close the connection? 891 Throwable exception = ((ConnectionError)command).getException(); 892 handleException(exception, null); 893 } 894 } 895 896 public ActiveMQMessage convertMessage(StompFrame command) throws IOException, JMSException { 897 ActiveMQMessage msg = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)).convertFrame(this, command); 898 return msg; 899 } 900 901 public StompFrame convertMessage(ActiveMQMessage message, boolean ignoreTransformation) throws IOException, JMSException { 902 if (ignoreTransformation == true) { 903 return frameTranslator.convertMessage(this, message); 904 } else { 905 FrameTranslator translator = findTranslator( 906 message.getStringProperty(Stomp.Headers.TRANSFORMATION), message.getDestination(), message.isAdvisory()); 907 return translator.convertMessage(this, message); 908 } 909 } 910 911 public StompTransport getStompTransport() { 912 return stompTransport; 913 } 914 915 public ActiveMQDestination createTempDestination(String name, boolean topic) { 916 ActiveMQDestination rc = tempDestinations.get(name); 917 if( rc == null ) { 918 if (topic) { 919 rc = new ActiveMQTempTopic(connectionId, tempDestinationGenerator.getNextSequenceId()); 920 } else { 921 rc = new ActiveMQTempQueue(connectionId, tempDestinationGenerator.getNextSequenceId()); 922 } 923 sendToActiveMQ(new DestinationInfo(connectionId, DestinationInfo.ADD_OPERATION_TYPE, rc), null); 924 tempDestinations.put(name, rc); 925 tempDestinationAmqToStompMap.put(rc.getQualifiedName(), name); 926 } 927 return rc; 928 } 929 930 public String getCreatedTempDestinationName(ActiveMQDestination destination) { 931 return tempDestinationAmqToStompMap.get(destination.getQualifiedName()); 932 } 933 934 public String getDefaultHeartBeat() { 935 return defaultHeartBeat; 936 } 937 938 public void setDefaultHeartBeat(String defaultHeartBeat) { 939 this.defaultHeartBeat = defaultHeartBeat; 940 } 941 942 /** 943 * @return the hbGracePeriodMultiplier 944 */ 945 public float getHbGracePeriodMultiplier() { 946 return hbGracePeriodMultiplier; 947 } 948 949 /** 950 * @param hbGracePeriodMultiplier the hbGracePeriodMultiplier to set 951 */ 952 public void setHbGracePeriodMultiplier(float hbGracePeriodMultiplier) { 953 this.hbGracePeriodMultiplier = hbGracePeriodMultiplier; 954 } 955 956 protected void configureInactivityMonitor(String heartBeatConfig) throws ProtocolException { 957 958 String[] keepAliveOpts = heartBeatConfig.split(Stomp.COMMA); 959 960 if (keepAliveOpts == null || keepAliveOpts.length != 2) { 961 throw new ProtocolException("Invalid heart-beat header:" + heartBeatConfig, true); 962 } else { 963 964 try { 965 hbReadInterval = (Long.parseLong(keepAliveOpts[0])); 966 hbWriteInterval = Long.parseLong(keepAliveOpts[1]); 967 } catch(NumberFormatException e) { 968 throw new ProtocolException("Invalid heart-beat header:" + heartBeatConfig, true); 969 } 970 971 try { 972 StompInactivityMonitor monitor = this.stompTransport.getInactivityMonitor(); 973 monitor.setReadCheckTime((long) (hbReadInterval * hbGracePeriodMultiplier)); 974 monitor.setInitialDelayTime(Math.min(hbReadInterval, hbWriteInterval)); 975 monitor.setWriteCheckTime(hbWriteInterval); 976 monitor.startMonitoring(); 977 } catch(Exception ex) { 978 hbReadInterval = 0; 979 hbWriteInterval = 0; 980 } 981 982 if (LOG.isDebugEnabled()) { 983 LOG.debug("Stomp Connect heartbeat conf RW[{},{}]", hbReadInterval, hbWriteInterval); 984 } 985 } 986 } 987 988 protected void sendReceipt(StompFrame command) { 989 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 990 if (receiptId != null) { 991 StompFrame sc = new StompFrame(); 992 sc.setAction(Stomp.Responses.RECEIPT); 993 sc.setHeaders(new HashMap<String, String>(1)); 994 sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 995 try { 996 sendToStomp(sc); 997 } catch (IOException e) { 998 LOG.warn("Could not send a receipt for {}", command, e); 999 } 1000 } 1001 } 1002 1003 /** 1004 * Retrieve the STOMP action value from a frame if the value is valid, otherwise 1005 * return an unknown string to allow for safe log output. 1006 * 1007 * @param command 1008 * The STOMP command to fetch an action from. 1009 * 1010 * @return the command action or a safe string to use in logging. 1011 */ 1012 protected Object safeGetAction(StompFrame command) { 1013 String result = "<Unknown>"; 1014 if (command != null && command.getAction() != null) { 1015 String action = command.getAction().trim(); 1016 1017 if (action != null) { 1018 switch (action) { 1019 case Stomp.Commands.SEND: 1020 case Stomp.Commands.ACK: 1021 case Stomp.Commands.NACK: 1022 case Stomp.Commands.BEGIN: 1023 case Stomp.Commands.COMMIT: 1024 case Stomp.Commands.ABORT: 1025 case Stomp.Commands.SUBSCRIBE: 1026 case Stomp.Commands.UNSUBSCRIBE: 1027 case Stomp.Commands.CONNECT: 1028 case Stomp.Commands.STOMP: 1029 case Stomp.Commands.DISCONNECT: 1030 result = action; 1031 break; 1032 default: 1033 break; 1034 } 1035 } 1036 } 1037 1038 return result; 1039 } 1040}