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 if (destination == null || "".equals(destination)) { 578 throw new ProtocolException("Invalid empty or 'null' Destination header"); 579 } 580 581 final ActiveMQDestination actualDest = translator.convertDestination(this, destination, true); 582 583 if (actualDest == null) { 584 throw new ProtocolException("Invalid 'null' Destination."); 585 } 586 587 final ConsumerId id = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId()); 588 ConsumerInfo consumerInfo = new ConsumerInfo(id); 589 consumerInfo.setPrefetchSize(actualDest.isQueue() ? 590 ActiveMQPrefetchPolicy.DEFAULT_QUEUE_PREFETCH : 591 headers.containsKey("activemq.subscriptionName") ? 592 ActiveMQPrefetchPolicy.DEFAULT_DURABLE_TOPIC_PREFETCH : ActiveMQPrefetchPolicy.DEFAULT_TOPIC_PREFETCH); 593 consumerInfo.setDispatchAsync(true); 594 595 String browser = headers.get(Stomp.Headers.Subscribe.BROWSER); 596 if (browser != null && browser.equals(Stomp.TRUE)) { 597 598 if (this.version.equals(Stomp.V1_0)) { 599 throw new ProtocolException("Queue Browser feature only valid for Stomp v1.1+ clients!"); 600 } 601 602 consumerInfo.setBrowser(true); 603 consumerInfo.setPrefetchSize(ActiveMQPrefetchPolicy.DEFAULT_QUEUE_BROWSER_PREFETCH); 604 } 605 606 String selector = headers.remove(Stomp.Headers.Subscribe.SELECTOR); 607 if (selector != null) { 608 consumerInfo.setSelector("convert_string_expressions:" + selector); 609 } 610 611 IntrospectionSupport.setProperties(consumerInfo, headers, "activemq."); 612 613 if (actualDest.isQueue() && consumerInfo.getSubscriptionName() != null) { 614 throw new ProtocolException("Invalid Subscription: cannot durably subscribe to a Queue destination!"); 615 } 616 617 consumerInfo.setDestination(actualDest); 618 619 StompSubscription stompSubscription; 620 if (!consumerInfo.isBrowser()) { 621 stompSubscription = new StompSubscription(this, subscriptionId, consumerInfo, headers.get(Stomp.Headers.TRANSFORMATION)); 622 } else { 623 stompSubscription = new StompQueueBrowserSubscription(this, subscriptionId, consumerInfo, headers.get(Stomp.Headers.TRANSFORMATION)); 624 } 625 stompSubscription.setDestination(actualDest); 626 627 String ackMode = headers.get(Stomp.Headers.Subscribe.ACK_MODE); 628 if (Stomp.Headers.Subscribe.AckModeValues.CLIENT.equals(ackMode)) { 629 stompSubscription.setAckMode(StompSubscription.CLIENT_ACK); 630 } else if (Stomp.Headers.Subscribe.AckModeValues.INDIVIDUAL.equals(ackMode)) { 631 stompSubscription.setAckMode(StompSubscription.INDIVIDUAL_ACK); 632 } else { 633 stompSubscription.setAckMode(StompSubscription.AUTO_ACK); 634 } 635 636 subscriptionsByConsumerId.put(id, stompSubscription); 637 // Stomp v1.0 doesn't need to set this header so we avoid an NPE if not set. 638 if (subscriptionId != null) { 639 subscriptions.put(subscriptionId, stompSubscription); 640 } 641 642 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 643 if (receiptId != null && consumerInfo.getPrefetchSize() > 0) { 644 645 final StompFrame cmd = command; 646 final int prefetch = consumerInfo.getPrefetchSize(); 647 648 // Since dispatch could beat the receipt we set prefetch to zero to start and then 649 // once we've sent our Receipt we are safe to turn on dispatch if the response isn't 650 // an error message. 651 consumerInfo.setPrefetchSize(0); 652 653 final ResponseHandler handler = new ResponseHandler() { 654 @Override 655 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 656 if (response.isException()) { 657 // Generally a command can fail.. but that does not invalidate the connection. 658 // We report back the failure but we don't close the connection. 659 Throwable exception = ((ExceptionResponse)response).getException(); 660 handleException(exception, cmd); 661 } else { 662 StompFrame sc = new StompFrame(); 663 sc.setAction(Stomp.Responses.RECEIPT); 664 sc.setHeaders(new HashMap<String, String>(1)); 665 sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 666 stompTransport.sendToStomp(sc); 667 668 ConsumerControl control = new ConsumerControl(); 669 control.setPrefetch(prefetch); 670 control.setDestination(actualDest); 671 control.setConsumerId(id); 672 673 sendToActiveMQ(control, null); 674 } 675 } 676 }; 677 678 sendToActiveMQ(consumerInfo, handler); 679 } else { 680 sendToActiveMQ(consumerInfo, createResponseHandler(command)); 681 } 682 } 683 684 protected void onStompUnsubscribe(StompFrame command) throws ProtocolException { 685 checkConnected(); 686 Map<String, String> headers = command.getHeaders(); 687 688 ActiveMQDestination destination = null; 689 Object o = headers.get(Stomp.Headers.Unsubscribe.DESTINATION); 690 if (o != null) { 691 destination = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)).convertDestination(this, (String)o, true); 692 } 693 694 String subscriptionId = headers.get(Stomp.Headers.Unsubscribe.ID); 695 if (!this.version.equals(Stomp.V1_0) && subscriptionId == null) { 696 throw new ProtocolException("UNSUBSCRIBE received without a subscription id!"); 697 } 698 699 if (subscriptionId == null && destination == null) { 700 throw new ProtocolException("Must specify the subscriptionId or the destination you are unsubscribing from"); 701 } 702 703 // check if it is a durable subscription 704 String durable = command.getHeaders().get("activemq.subscriptionName"); 705 String clientId = durable; 706 if (!this.version.equals(Stomp.V1_0)) { 707 clientId = connectionInfo.getClientId(); 708 } 709 710 if (durable != null) { 711 RemoveSubscriptionInfo info = new RemoveSubscriptionInfo(); 712 info.setClientId(clientId); 713 info.setSubscriptionName(durable); 714 info.setConnectionId(connectionId); 715 sendToActiveMQ(info, createResponseHandler(command)); 716 return; 717 } 718 719 if (subscriptionId != null) { 720 StompSubscription sub = this.subscriptions.remove(subscriptionId); 721 if (sub != null) { 722 sendToActiveMQ(sub.getConsumerInfo().createRemoveCommand(), createResponseHandler(command)); 723 return; 724 } 725 } else { 726 // Unsubscribing using a destination is a bit weird if multiple subscriptions 727 // are created with the same destination. 728 for (Iterator<StompSubscription> iter = subscriptionsByConsumerId.values().iterator(); iter.hasNext();) { 729 StompSubscription sub = iter.next(); 730 if (destination != null && destination.equals(sub.getDestination())) { 731 sendToActiveMQ(sub.getConsumerInfo().createRemoveCommand(), createResponseHandler(command)); 732 iter.remove(); 733 return; 734 } 735 } 736 } 737 738 throw new ProtocolException("No subscription matched."); 739 } 740 741 ConnectionInfo connectionInfo = new ConnectionInfo(); 742 743 protected void onStompConnect(final StompFrame command) throws ProtocolException { 744 745 if (connected.get()) { 746 throw new ProtocolException("Already connected."); 747 } 748 749 final Map<String, String> headers = command.getHeaders(); 750 751 // allow anyone to login for now 752 String login = headers.get(Stomp.Headers.Connect.LOGIN); 753 String passcode = headers.get(Stomp.Headers.Connect.PASSCODE); 754 String clientId = headers.get(Stomp.Headers.Connect.CLIENT_ID); 755 String heartBeat = headers.get(Stomp.Headers.Connect.HEART_BEAT); 756 757 if (heartBeat == null) { 758 heartBeat = defaultHeartBeat; 759 } 760 761 this.version = StompCodec.detectVersion(headers); 762 763 configureInactivityMonitor(heartBeat.trim()); 764 765 IntrospectionSupport.setProperties(connectionInfo, headers, "activemq."); 766 connectionInfo.setConnectionId(connectionId); 767 if (clientId != null) { 768 connectionInfo.setClientId(clientId); 769 } else { 770 connectionInfo.setClientId("" + connectionInfo.getConnectionId().toString()); 771 } 772 773 connectionInfo.setResponseRequired(true); 774 connectionInfo.setUserName(login); 775 connectionInfo.setPassword(passcode); 776 connectionInfo.setTransportContext(command.getTransportContext()); 777 778 sendToActiveMQ(connectionInfo, new ResponseHandler() { 779 @Override 780 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 781 782 if (response.isException()) { 783 // If the connection attempt fails we close the socket. 784 Throwable exception = ((ExceptionResponse)response).getException(); 785 handleException(exception, command); 786 getStompTransport().onException(IOExceptionSupport.create(exception)); 787 return; 788 } 789 790 final SessionInfo sessionInfo = new SessionInfo(sessionId); 791 sendToActiveMQ(sessionInfo, null); 792 793 final ProducerInfo producerInfo = new ProducerInfo(producerId); 794 sendToActiveMQ(producerInfo, new ResponseHandler() { 795 @Override 796 public void onResponse(ProtocolConverter converter, Response response) throws IOException { 797 798 if (response.isException()) { 799 // If the connection attempt fails we close the socket. 800 Throwable exception = ((ExceptionResponse)response).getException(); 801 handleException(exception, command); 802 getStompTransport().onException(IOExceptionSupport.create(exception)); 803 } 804 805 connected.set(true); 806 HashMap<String, String> responseHeaders = new HashMap<String, String>(); 807 808 responseHeaders.put(Stomp.Headers.Connected.SESSION, connectionInfo.getClientId()); 809 String requestId = headers.get(Stomp.Headers.Connect.REQUEST_ID); 810 if (requestId == null) { 811 // TODO legacy 812 requestId = headers.get(Stomp.Headers.RECEIPT_REQUESTED); 813 } 814 if (requestId != null) { 815 // TODO legacy 816 responseHeaders.put(Stomp.Headers.Connected.RESPONSE_ID, requestId); 817 responseHeaders.put(Stomp.Headers.Response.RECEIPT_ID, requestId); 818 } 819 820 responseHeaders.put(Stomp.Headers.Connected.VERSION, version); 821 responseHeaders.put(Stomp.Headers.Connected.HEART_BEAT, 822 String.format("%d,%d", hbWriteInterval, hbReadInterval)); 823 responseHeaders.put(Stomp.Headers.Connected.SERVER, "ActiveMQ/"+BROKER_VERSION); 824 825 StompFrame sc = new StompFrame(); 826 sc.setAction(Stomp.Responses.CONNECTED); 827 sc.setHeaders(responseHeaders); 828 sendToStomp(sc); 829 830 StompWireFormat format = stompTransport.getWireFormat(); 831 if (format != null) { 832 format.setStompVersion(version); 833 } 834 } 835 }); 836 } 837 }); 838 } 839 840 protected void onStompDisconnect(StompFrame command) throws ProtocolException { 841 if (connected.get()) { 842 sendToActiveMQ(connectionInfo.createRemoveCommand(), createResponseHandler(command)); 843 sendToActiveMQ(new ShutdownInfo(), createResponseHandler(command)); 844 connected.set(false); 845 } 846 } 847 848 protected void checkConnected() throws ProtocolException { 849 if (!connected.get()) { 850 throw new ProtocolException("Not connected."); 851 } 852 } 853 854 /** 855 * Dispatch a ActiveMQ command 856 * 857 * @param command 858 * @throws IOException 859 */ 860 public void onActiveMQCommand(Command command) throws IOException, JMSException { 861 if (command.isResponse()) { 862 Response response = (Response)command; 863 ResponseHandler rh = resposeHandlers.remove(Integer.valueOf(response.getCorrelationId())); 864 if (rh != null) { 865 rh.onResponse(this, response); 866 } else { 867 // Pass down any unexpected errors. Should this close the connection? 868 if (response.isException()) { 869 Throwable exception = ((ExceptionResponse)response).getException(); 870 handleException(exception, null); 871 } 872 } 873 } else if (command.isMessageDispatch()) { 874 MessageDispatch md = (MessageDispatch)command; 875 StompSubscription sub = subscriptionsByConsumerId.get(md.getConsumerId()); 876 if (sub != null) { 877 String ackId = null; 878 if (version.equals(Stomp.V1_2) && sub.getAckMode() != Stomp.Headers.Subscribe.AckModeValues.AUTO && md.getMessage() != null) { 879 AckEntry pendingAck = new AckEntry(md.getMessage().getMessageId().toString(), sub); 880 ackId = this.ACK_ID_GENERATOR.generateId(); 881 this.pedingAcks.put(ackId, pendingAck); 882 } 883 try { 884 sub.onMessageDispatch(md, ackId); 885 } catch (Exception ex) { 886 if (ackId != null) { 887 this.pedingAcks.remove(ackId); 888 } 889 } 890 } 891 } else if (command.getDataStructureType() == CommandTypes.KEEP_ALIVE_INFO) { 892 stompTransport.sendToStomp(ping); 893 } else if (command.getDataStructureType() == ConnectionError.DATA_STRUCTURE_TYPE) { 894 // Pass down any unexpected async errors. Should this close the connection? 895 Throwable exception = ((ConnectionError)command).getException(); 896 handleException(exception, null); 897 } 898 } 899 900 public ActiveMQMessage convertMessage(StompFrame command) throws IOException, JMSException { 901 ActiveMQMessage msg = findTranslator(command.getHeaders().get(Stomp.Headers.TRANSFORMATION)).convertFrame(this, command); 902 return msg; 903 } 904 905 public StompFrame convertMessage(ActiveMQMessage message, boolean ignoreTransformation) throws IOException, JMSException { 906 if (ignoreTransformation == true) { 907 return frameTranslator.convertMessage(this, message); 908 } else { 909 FrameTranslator translator = findTranslator( 910 message.getStringProperty(Stomp.Headers.TRANSFORMATION), message.getDestination(), message.isAdvisory()); 911 return translator.convertMessage(this, message); 912 } 913 } 914 915 public StompTransport getStompTransport() { 916 return stompTransport; 917 } 918 919 public ActiveMQDestination createTempDestination(String name, boolean topic) { 920 ActiveMQDestination rc = tempDestinations.get(name); 921 if( rc == null ) { 922 if (topic) { 923 rc = new ActiveMQTempTopic(connectionId, tempDestinationGenerator.getNextSequenceId()); 924 } else { 925 rc = new ActiveMQTempQueue(connectionId, tempDestinationGenerator.getNextSequenceId()); 926 } 927 sendToActiveMQ(new DestinationInfo(connectionId, DestinationInfo.ADD_OPERATION_TYPE, rc), null); 928 tempDestinations.put(name, rc); 929 tempDestinationAmqToStompMap.put(rc.getQualifiedName(), name); 930 } 931 return rc; 932 } 933 934 public String getCreatedTempDestinationName(ActiveMQDestination destination) { 935 return tempDestinationAmqToStompMap.get(destination.getQualifiedName()); 936 } 937 938 public String getDefaultHeartBeat() { 939 return defaultHeartBeat; 940 } 941 942 public void setDefaultHeartBeat(String defaultHeartBeat) { 943 this.defaultHeartBeat = defaultHeartBeat; 944 } 945 946 /** 947 * @return the hbGracePeriodMultiplier 948 */ 949 public float getHbGracePeriodMultiplier() { 950 return hbGracePeriodMultiplier; 951 } 952 953 /** 954 * @param hbGracePeriodMultiplier the hbGracePeriodMultiplier to set 955 */ 956 public void setHbGracePeriodMultiplier(float hbGracePeriodMultiplier) { 957 this.hbGracePeriodMultiplier = hbGracePeriodMultiplier; 958 } 959 960 protected void configureInactivityMonitor(String heartBeatConfig) throws ProtocolException { 961 962 String[] keepAliveOpts = heartBeatConfig.split(Stomp.COMMA); 963 964 if (keepAliveOpts == null || keepAliveOpts.length != 2) { 965 throw new ProtocolException("Invalid heart-beat header:" + heartBeatConfig, true); 966 } else { 967 968 try { 969 hbReadInterval = (Long.parseLong(keepAliveOpts[0])); 970 hbWriteInterval = Long.parseLong(keepAliveOpts[1]); 971 } catch(NumberFormatException e) { 972 throw new ProtocolException("Invalid heart-beat header:" + heartBeatConfig, true); 973 } 974 975 try { 976 StompInactivityMonitor monitor = this.stompTransport.getInactivityMonitor(); 977 monitor.setReadCheckTime((long) (hbReadInterval * hbGracePeriodMultiplier)); 978 monitor.setInitialDelayTime(Math.min(hbReadInterval, hbWriteInterval)); 979 monitor.setWriteCheckTime(hbWriteInterval); 980 monitor.startMonitoring(); 981 } catch(Exception ex) { 982 hbReadInterval = 0; 983 hbWriteInterval = 0; 984 } 985 986 if (LOG.isDebugEnabled()) { 987 LOG.debug("Stomp Connect heartbeat conf RW[{},{}]", hbReadInterval, hbWriteInterval); 988 } 989 } 990 } 991 992 protected void sendReceipt(StompFrame command) { 993 final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED); 994 if (receiptId != null) { 995 StompFrame sc = new StompFrame(); 996 sc.setAction(Stomp.Responses.RECEIPT); 997 sc.setHeaders(new HashMap<String, String>(1)); 998 sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId); 999 try { 1000 sendToStomp(sc); 1001 } catch (IOException e) { 1002 LOG.warn("Could not send a receipt for {}", command, e); 1003 } 1004 } 1005 } 1006 1007 /** 1008 * Retrieve the STOMP action value from a frame if the value is valid, otherwise 1009 * return an unknown string to allow for safe log output. 1010 * 1011 * @param command 1012 * The STOMP command to fetch an action from. 1013 * 1014 * @return the command action or a safe string to use in logging. 1015 */ 1016 protected Object safeGetAction(StompFrame command) { 1017 String result = "<Unknown>"; 1018 if (command != null && command.getAction() != null) { 1019 String action = command.getAction().trim(); 1020 1021 if (action != null) { 1022 switch (action) { 1023 case Stomp.Commands.SEND: 1024 case Stomp.Commands.ACK: 1025 case Stomp.Commands.NACK: 1026 case Stomp.Commands.BEGIN: 1027 case Stomp.Commands.COMMIT: 1028 case Stomp.Commands.ABORT: 1029 case Stomp.Commands.SUBSCRIBE: 1030 case Stomp.Commands.UNSUBSCRIBE: 1031 case Stomp.Commands.CONNECT: 1032 case Stomp.Commands.STOMP: 1033 case Stomp.Commands.DISCONNECT: 1034 result = action; 1035 break; 1036 default: 1037 break; 1038 } 1039 } 1040 } 1041 1042 return result; 1043 } 1044}