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     */
017    package org.apache.activemq.transport.ws;
018    
019    import org.apache.activemq.command.Command;
020    import org.apache.activemq.transport.TransportSupport;
021    import org.apache.activemq.transport.mqtt.MQTTInactivityMonitor;
022    import org.apache.activemq.transport.mqtt.MQTTProtocolConverter;
023    import org.apache.activemq.transport.mqtt.MQTTTransport;
024    import org.apache.activemq.transport.mqtt.MQTTWireFormat;
025    import org.apache.activemq.util.ByteSequence;
026    import org.apache.activemq.util.IOExceptionSupport;
027    import org.apache.activemq.util.ServiceStopper;
028    import org.eclipse.jetty.websocket.WebSocket;
029    import org.fusesource.mqtt.codec.DISCONNECT;
030    import org.fusesource.mqtt.codec.MQTTFrame;
031    import org.slf4j.Logger;
032    import org.slf4j.LoggerFactory;
033    
034    import java.io.IOException;
035    import java.security.cert.X509Certificate;
036    import java.util.concurrent.CountDownLatch;
037    
038    public class MQTTSocket  extends TransportSupport implements WebSocket.OnBinaryMessage, MQTTTransport {
039    
040        private static final Logger LOG = LoggerFactory.getLogger(MQTTSocket.class);
041        Connection outbound;
042        MQTTProtocolConverter protocolConverter = new MQTTProtocolConverter(this, null);
043        MQTTWireFormat wireFormat = new MQTTWireFormat();
044        private final CountDownLatch socketTransportStarted = new CountDownLatch(1);
045    
046        @Override
047        public void onMessage(byte[] bytes, int offset, int length) {
048            if (!transportStartedAtLeastOnce()) {
049                LOG.debug("Waiting for StompSocket to be properly started...");
050                try {
051                    socketTransportStarted.await();
052                } catch (InterruptedException e) {
053                    LOG.warn("While waiting for StompSocket to be properly started, we got interrupted!! Should be okay, but you could see race conditions...");
054                }
055            }
056    
057            try {
058                MQTTFrame frame = (MQTTFrame)wireFormat.unmarshal(new ByteSequence(bytes, offset, length));
059                protocolConverter.onMQTTCommand(frame);
060            } catch (Exception e) {
061                onException(IOExceptionSupport.create(e));
062            }
063        }
064    
065        @Override
066        public void onOpen(Connection connection) {
067            this.outbound = connection;
068        }
069    
070        @Override
071        public void onClose(int closeCode, String message) {
072            try {
073                protocolConverter.onMQTTCommand(new DISCONNECT().encode());
074            } catch (Exception e) {
075                LOG.warn("Failed to close WebSocket", e);
076            }
077        }
078    
079        protected void doStart() throws Exception {
080            socketTransportStarted.countDown();
081        }
082    
083        @Override
084        protected void doStop(ServiceStopper stopper) throws Exception {
085        }
086    
087        private boolean transportStartedAtLeastOnce() {
088            return socketTransportStarted.getCount() == 0;
089        }
090    
091        @Override
092        public int getReceiveCounter() {
093            return 0;
094        }
095    
096        @Override
097        public String getRemoteAddress() {
098            return "MQTTSocket_" + this.hashCode();
099        }
100    
101        @Override
102        public void oneway(Object command) throws IOException {
103            try {
104                protocolConverter.onActiveMQCommand((Command)command);
105            } catch (Exception e) {
106                onException(IOExceptionSupport.create(e));
107            }
108        }
109    
110        @Override
111        public void sendToActiveMQ(Command command) {
112            doConsume(command);
113        }
114    
115        @Override
116        public void sendToMQTT(MQTTFrame command) throws IOException {
117            ByteSequence bytes = wireFormat.marshal(command);
118            outbound.sendMessage(bytes.getData(), 0, bytes.getLength());
119        }
120    
121        @Override
122        public X509Certificate[] getPeerCertificates() {
123            return new X509Certificate[0];
124        }
125    
126        @Override
127        public MQTTInactivityMonitor getInactivityMonitor() {
128            return null;
129        }
130    
131        @Override
132        public MQTTWireFormat getWireFormat() {
133            return wireFormat;
134        }
135    }