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.ws; 018 019import java.io.IOException; 020import java.util.concurrent.CountDownLatch; 021 022import org.apache.activemq.command.Command; 023import org.apache.activemq.command.KeepAliveInfo; 024import org.apache.activemq.transport.TransportSupport; 025import org.apache.activemq.transport.stomp.ProtocolConverter; 026import org.apache.activemq.transport.stomp.StompFrame; 027import org.apache.activemq.transport.stomp.StompInactivityMonitor; 028import org.apache.activemq.transport.stomp.StompTransport; 029import org.apache.activemq.transport.stomp.StompWireFormat; 030import org.apache.activemq.util.ByteSequence; 031import org.apache.activemq.util.IOExceptionSupport; 032import org.apache.activemq.util.ServiceStopper; 033import org.slf4j.Logger; 034import org.slf4j.LoggerFactory; 035 036/** 037 * Base implementation of a STOMP based WebSocket handler. 038 */ 039public abstract class AbstractStompSocket extends TransportSupport implements StompTransport { 040 041 private static final Logger LOG = LoggerFactory.getLogger(AbstractStompSocket.class); 042 043 protected ProtocolConverter protocolConverter = new ProtocolConverter(this, null); 044 protected StompWireFormat wireFormat = new StompWireFormat(); 045 protected final CountDownLatch socketTransportStarted = new CountDownLatch(1); 046 protected final StompInactivityMonitor stompInactivityMonitor = new StompInactivityMonitor(this, wireFormat); 047 protected volatile int receiveCounter; 048 049 @Override 050 public void oneway(Object command) throws IOException { 051 try { 052 protocolConverter.onActiveMQCommand((Command)command); 053 } catch (Exception e) { 054 onException(IOExceptionSupport.create(e)); 055 } 056 } 057 058 @Override 059 public void sendToActiveMQ(Command command) { 060 doConsume(command); 061 } 062 063 @Override 064 protected void doStop(ServiceStopper stopper) throws Exception { 065 stompInactivityMonitor.stop(); 066 handleStopped(); 067 } 068 069 @Override 070 protected void doStart() throws Exception { 071 socketTransportStarted.countDown(); 072 stompInactivityMonitor.setTransportListener(getTransportListener()); 073 stompInactivityMonitor.startConnectCheckTask(); 074 } 075 076 //----- Abstract methods for subclasses to implement ---------------------// 077 078 @Override 079 public abstract void sendToStomp(StompFrame command) throws IOException; 080 081 /** 082 * Called when the transport is stopping to allow the dervied classes 083 * a chance to close WebSocket resources. 084 * 085 * @throws IOException if an error occurs during the stop. 086 */ 087 public abstract void handleStopped() throws IOException; 088 089 //----- Accessor methods -------------------------------------------------// 090 091 @Override 092 public StompInactivityMonitor getInactivityMonitor() { 093 return stompInactivityMonitor; 094 } 095 096 @Override 097 public StompWireFormat getWireFormat() { 098 return wireFormat; 099 } 100 101 @Override 102 public String getRemoteAddress() { 103 return "StompSocket_" + this.hashCode(); 104 } 105 106 @Override 107 public int getReceiveCounter() { 108 return receiveCounter; 109 } 110 111 //----- Internal implementation ------------------------------------------// 112 113 protected void processStompFrame(String data) { 114 115 if (!transportStartedAtLeastOnce()) { 116 LOG.debug("Waiting for StompSocket to be properly started..."); 117 try { 118 socketTransportStarted.await(); 119 } catch (InterruptedException e) { 120 LOG.warn("While waiting for StompSocket to be properly started, we got interrupted!! Should be okay, but you could see race conditions..."); 121 } 122 } 123 124 try { 125 if (data != null) { 126 receiveCounter += data.length(); 127 128 if (data.equals("\n")) { 129 stompInactivityMonitor.onCommand(new KeepAliveInfo()); 130 } else { 131 protocolConverter.onStompCommand((StompFrame)wireFormat.unmarshal(new ByteSequence(data.getBytes("UTF-8")))); 132 } 133 } 134 } catch (Exception e) { 135 onException(IOExceptionSupport.create(e)); 136 } 137 } 138 139 private boolean transportStartedAtLeastOnce() { 140 return socketTransportStarted.getCount() == 0; 141 } 142}