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.mqtt.strategy; 018 019import java.io.IOException; 020import java.util.ArrayList; 021import java.util.Collections; 022import java.util.HashSet; 023import java.util.List; 024import java.util.Set; 025import java.util.concurrent.ConcurrentHashMap; 026import java.util.concurrent.ConcurrentMap; 027 028import org.apache.activemq.broker.BrokerService; 029import org.apache.activemq.broker.BrokerServiceAware; 030import org.apache.activemq.broker.ConnectionContext; 031import org.apache.activemq.broker.region.DurableTopicSubscription; 032import org.apache.activemq.broker.region.PrefetchSubscription; 033import org.apache.activemq.broker.region.RegionBroker; 034import org.apache.activemq.broker.region.Subscription; 035import org.apache.activemq.broker.region.TopicRegion; 036import org.apache.activemq.broker.region.virtual.VirtualTopicInterceptor; 037import org.apache.activemq.command.ActiveMQDestination; 038import org.apache.activemq.command.ActiveMQTopic; 039import org.apache.activemq.command.ConsumerId; 040import org.apache.activemq.command.ConsumerInfo; 041import org.apache.activemq.command.ExceptionResponse; 042import org.apache.activemq.command.RemoveInfo; 043import org.apache.activemq.command.RemoveSubscriptionInfo; 044import org.apache.activemq.command.Response; 045import org.apache.activemq.command.SubscriptionInfo; 046import org.apache.activemq.transport.mqtt.MQTTProtocolConverter; 047import org.apache.activemq.transport.mqtt.MQTTProtocolException; 048import org.apache.activemq.transport.mqtt.MQTTProtocolSupport; 049import org.apache.activemq.transport.mqtt.MQTTSubscription; 050import org.apache.activemq.transport.mqtt.ResponseHandler; 051import org.apache.activemq.util.LongSequenceGenerator; 052import org.fusesource.mqtt.client.QoS; 053import org.fusesource.mqtt.client.Topic; 054import org.slf4j.Logger; 055import org.slf4j.LoggerFactory; 056 057/** 058 * Abstract implementation of the {@link MQTTSubscriptionStrategy} interface providing 059 * the base functionality that is common to most implementations. 060 */ 061public abstract class AbstractMQTTSubscriptionStrategy implements MQTTSubscriptionStrategy, BrokerServiceAware { 062 063 private static final Logger LOG = LoggerFactory.getLogger(AbstractMQTTSubscriptionStrategy.class); 064 065 private static final byte SUBSCRIBE_ERROR = (byte) 0x80; 066 067 protected MQTTProtocolConverter protocol; 068 protected BrokerService brokerService; 069 070 protected final ConcurrentMap<ConsumerId, MQTTSubscription> subscriptionsByConsumerId = new ConcurrentHashMap<ConsumerId, MQTTSubscription>(); 071 protected final ConcurrentMap<String, MQTTSubscription> mqttSubscriptionByTopic = new ConcurrentHashMap<String, MQTTSubscription>(); 072 protected final Set<String> restoredDurableSubs = Collections.synchronizedSet(new HashSet<String>()); 073 074 protected final LongSequenceGenerator consumerIdGenerator = new LongSequenceGenerator(); 075 076 @Override 077 public void initialize(MQTTProtocolConverter protocol) throws MQTTProtocolException { 078 setProtocolConverter(protocol); 079 } 080 081 @Override 082 public void setBrokerService(BrokerService brokerService) { 083 this.brokerService = brokerService; 084 } 085 086 @Override 087 public void setProtocolConverter(MQTTProtocolConverter parent) { 088 this.protocol = parent; 089 } 090 091 @Override 092 public MQTTProtocolConverter getProtocolConverter() { 093 return protocol; 094 } 095 096 @Override 097 public byte onSubscribe(final Topic topic) throws MQTTProtocolException { 098 099 final String destinationName = topic.name().toString(); 100 final QoS requestedQoS = topic.qos(); 101 102 final MQTTSubscription mqttSubscription = mqttSubscriptionByTopic.get(destinationName); 103 if (mqttSubscription != null) { 104 if (requestedQoS != mqttSubscription.getQoS()) { 105 // remove old subscription as the QoS has changed 106 onUnSubscribe(destinationName); 107 } else { 108 try { 109 onReSubscribe(mqttSubscription); 110 } catch (IOException e) { 111 throw new MQTTProtocolException("Failed to find subscription strategy", true, e); 112 } 113 return (byte) requestedQoS.ordinal(); 114 } 115 } 116 117 try { 118 return onSubscribe(destinationName, requestedQoS); 119 } catch (IOException e) { 120 throw new MQTTProtocolException("Failed while intercepting subscribe", true, e); 121 } 122 } 123 124 @Override 125 public void onReSubscribe(MQTTSubscription mqttSubscription) throws MQTTProtocolException { 126 String topicName = mqttSubscription.getTopicName(); 127 128 // get TopicRegion 129 RegionBroker regionBroker; 130 try { 131 regionBroker = (RegionBroker) brokerService.getBroker().getAdaptor(RegionBroker.class); 132 } catch (Exception e) { 133 throw new MQTTProtocolException("Error subscribing to " + topicName + ": " + e.getMessage(), false, e); 134 } 135 final TopicRegion topicRegion = (TopicRegion) regionBroker.getTopicRegion(); 136 137 final ConsumerInfo consumerInfo = mqttSubscription.getConsumerInfo(); 138 final ConsumerId consumerId = consumerInfo.getConsumerId(); 139 140 // use actual client id used to create connection to lookup connection 141 // context 142 String connectionInfoClientId = protocol.getClientId(); 143 // for zero-byte client ids we used connection id 144 if (connectionInfoClientId == null || connectionInfoClientId.isEmpty()) { 145 connectionInfoClientId = protocol.getConnectionId().toString(); 146 } 147 final ConnectionContext connectionContext = regionBroker.getConnectionContext(connectionInfoClientId); 148 149 // get all matching Topics 150 final Set<org.apache.activemq.broker.region.Destination> matchingDestinations = 151 topicRegion.getDestinations(mqttSubscription.getDestination()); 152 for (org.apache.activemq.broker.region.Destination dest : matchingDestinations) { 153 154 // recover retroactive messages for matching subscription 155 for (Subscription subscription : dest.getConsumers()) { 156 if (subscription.getConsumerInfo().getConsumerId().equals(consumerId)) { 157 try { 158 if (dest instanceof org.apache.activemq.broker.region.Topic) { 159 ((org.apache.activemq.broker.region.Topic) dest).recoverRetroactiveMessages(connectionContext, subscription); 160 } else if (dest instanceof VirtualTopicInterceptor) { 161 ((VirtualTopicInterceptor) dest).getTopic().recoverRetroactiveMessages(connectionContext, subscription); 162 } 163 if (subscription instanceof PrefetchSubscription) { 164 // request dispatch for prefetch subs 165 PrefetchSubscription prefetchSubscription = (PrefetchSubscription) subscription; 166 prefetchSubscription.dispatchPending(); 167 } 168 } catch (Exception e) { 169 throw new MQTTProtocolException("Error recovering retained messages for " + dest.getName() + ": " + e.getMessage(), false, e); 170 } 171 break; 172 } 173 } 174 } 175 } 176 177 @Override 178 public ActiveMQDestination onSend(String topicName) { 179 return new ActiveMQTopic(topicName); 180 } 181 182 @Override 183 public String onSend(ActiveMQDestination destination) { 184 return destination.getPhysicalName(); 185 } 186 187 @Override 188 public boolean isControlTopic(ActiveMQDestination destination) { 189 return destination.getPhysicalName().startsWith("$"); 190 } 191 192 @Override 193 public MQTTSubscription getSubscription(ConsumerId consumerId) { 194 return subscriptionsByConsumerId.get(consumerId); 195 } 196 197 protected ConsumerId getNextConsumerId() { 198 return new ConsumerId(protocol.getSessionId(), consumerIdGenerator.getNextSequenceId()); 199 } 200 201 protected byte doSubscribe(ConsumerInfo consumerInfo, final String topicName, final QoS qoS) throws MQTTProtocolException { 202 203 MQTTSubscription mqttSubscription = new MQTTSubscription(protocol, topicName, qoS, consumerInfo); 204 205 // optimistic add to local maps first to be able to handle commands in onActiveMQCommand 206 subscriptionsByConsumerId.put(consumerInfo.getConsumerId(), mqttSubscription); 207 mqttSubscriptionByTopic.put(topicName, mqttSubscription); 208 209 final byte[] qos = {-1}; 210 protocol.sendToActiveMQ(consumerInfo, new ResponseHandler() { 211 @Override 212 public void onResponse(MQTTProtocolConverter converter, Response response) throws IOException { 213 // validate subscription request 214 if (response.isException()) { 215 final Throwable throwable = ((ExceptionResponse) response).getException(); 216 LOG.warn("Error subscribing to {}", topicName, throwable); 217 // version 3.1 don't supports silent fail 218 // version 3.1.1 send "error" qos 219 if (protocol.version == MQTTProtocolConverter.V3_1_1) { 220 qos[0] = SUBSCRIBE_ERROR; 221 } else { 222 qos[0] = (byte) qoS.ordinal(); 223 } 224 } else { 225 qos[0] = (byte) qoS.ordinal(); 226 } 227 } 228 }); 229 230 if (qos[0] == SUBSCRIBE_ERROR) { 231 // remove from local maps if subscribe failed 232 subscriptionsByConsumerId.remove(consumerInfo.getConsumerId()); 233 mqttSubscriptionByTopic.remove(topicName); 234 } 235 236 return qos[0]; 237 } 238 239 public void doUnSubscribe(MQTTSubscription subscription) { 240 mqttSubscriptionByTopic.remove(subscription.getTopicName()); 241 ConsumerInfo info = subscription.getConsumerInfo(); 242 if (info != null) { 243 subscriptionsByConsumerId.remove(info.getConsumerId()); 244 245 RemoveInfo removeInfo = info.createRemoveCommand(); 246 protocol.sendToActiveMQ(removeInfo, new ResponseHandler() { 247 @Override 248 public void onResponse(MQTTProtocolConverter converter, Response response) throws IOException { 249 // ignore failures.. 250 } 251 }); 252 } 253 } 254 255 //----- Durable Subscription management methods --------------------------// 256 257 protected void deleteDurableSubs(List<SubscriptionInfo> subs) { 258 try { 259 for (SubscriptionInfo sub : subs) { 260 RemoveSubscriptionInfo rsi = new RemoveSubscriptionInfo(); 261 rsi.setConnectionId(protocol.getConnectionId()); 262 rsi.setSubscriptionName(sub.getSubcriptionName()); 263 rsi.setClientId(sub.getClientId()); 264 protocol.sendToActiveMQ(rsi, new ResponseHandler() { 265 @Override 266 public void onResponse(MQTTProtocolConverter converter, Response response) throws IOException { 267 // ignore failures.. 268 } 269 }); 270 } 271 } catch (Throwable e) { 272 LOG.warn("Could not delete the MQTT durable subs.", e); 273 } 274 } 275 276 protected void restoreDurableSubs(List<SubscriptionInfo> subs) { 277 try { 278 for (SubscriptionInfo sub : subs) { 279 String name = sub.getSubcriptionName(); 280 String[] split = name.split(":", 2); 281 QoS qoS = QoS.valueOf(split[0]); 282 onSubscribe(new Topic(split[1], qoS)); 283 // mark this durable subscription as restored by Broker 284 restoredDurableSubs.add(MQTTProtocolSupport.convertMQTTToActiveMQ(split[1])); 285 } 286 } catch (IOException e) { 287 LOG.warn("Could not restore the MQTT durable subs.", e); 288 } 289 } 290 291 protected List<SubscriptionInfo> lookupSubscription(String clientId) throws MQTTProtocolException { 292 List<SubscriptionInfo> result = new ArrayList<SubscriptionInfo>(); 293 RegionBroker regionBroker; 294 295 try { 296 regionBroker = (RegionBroker) brokerService.getBroker().getAdaptor(RegionBroker.class); 297 } catch (Exception e) { 298 throw new MQTTProtocolException("Error recovering durable subscriptions: " + e.getMessage(), false, e); 299 } 300 301 final TopicRegion topicRegion = (TopicRegion) regionBroker.getTopicRegion(); 302 List<DurableTopicSubscription> subscriptions = topicRegion.lookupSubscriptions(clientId); 303 if (subscriptions != null) { 304 for (DurableTopicSubscription subscription : subscriptions) { 305 LOG.debug("Recovered durable sub:{} on connect", subscription); 306 307 SubscriptionInfo info = new SubscriptionInfo(); 308 309 info.setDestination(subscription.getActiveMQDestination()); 310 info.setSubcriptionName(subscription.getSubscriptionKey().getSubscriptionName()); 311 info.setClientId(clientId); 312 313 result.add(info); 314 } 315 } 316 317 return result; 318 } 319}