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}