forked from apache/activemq
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
https://issues.apache.org/jira/browse/AMQ-6339
Add support for AMQP client to connect using WebSockets.
- Loading branch information
Showing
53 changed files
with
2,228 additions
and
562 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
143 changes: 143 additions & 0 deletions
143
activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWSTransport.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.activemq.transport.amqp; | ||
|
|
||
| import java.io.IOException; | ||
| import java.net.URI; | ||
| import java.nio.ByteBuffer; | ||
| import java.security.cert.X509Certificate; | ||
|
|
||
| import org.apache.activemq.transport.TransportSupport; | ||
| import org.apache.activemq.transport.amqp.AmqpFrameParser.AMQPFrameSink; | ||
| import org.apache.activemq.transport.ws.WSTransport; | ||
| import org.apache.activemq.util.IOExceptionSupport; | ||
| import org.apache.activemq.util.ServiceStopper; | ||
| import org.apache.activemq.wireformat.WireFormat; | ||
|
|
||
| /** | ||
| * An AMQP based WebSocket transport implementation. | ||
| */ | ||
| public class AmqpWSTransport extends TransportSupport implements WSTransport, AMQPFrameSink { | ||
|
|
||
| private final AmqpFrameParser frameReader = new AmqpFrameParser(this); | ||
| private final URI remoteLocation; | ||
|
|
||
| private WSTransportSink outputSink; | ||
| private int receiveCounter; | ||
| private X509Certificate[] certificates; | ||
|
|
||
| /** | ||
| * Create a new Transport instance. | ||
| * | ||
| * @param location | ||
| * the remote location where the client connection is from. | ||
| * @param wireFormat | ||
| * the WireFormat instance that configures this Transport. | ||
| */ | ||
| public AmqpWSTransport(URI location, WireFormat wireFormat) { | ||
| super(); | ||
|
|
||
| remoteLocation = location; | ||
| frameReader.setWireFormat((AmqpWireFormat) wireFormat); | ||
| } | ||
|
|
||
| @Override | ||
| public void setTransportSink(WSTransportSink outputSink) { | ||
| this.outputSink = outputSink; | ||
| } | ||
|
|
||
| @Override | ||
| public void oneway(Object command) throws IOException { | ||
| if (command instanceof ByteBuffer) { | ||
| outputSink.onSocketOutboundBinary((ByteBuffer) command); | ||
| } else { | ||
| throw new IOException("Unexpected output command."); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String getRemoteAddress() { | ||
| return remoteLocation.toASCIIString(); | ||
| } | ||
|
|
||
| @Override | ||
| public int getReceiveCounter() { | ||
| return receiveCounter; | ||
| } | ||
|
|
||
| @Override | ||
| public X509Certificate[] getPeerCertificates() { | ||
| return certificates; | ||
| } | ||
|
|
||
| @Override | ||
| public void setPeerCertificates(X509Certificate[] certificates) { | ||
| this.certificates = certificates; | ||
| } | ||
|
|
||
| @Override | ||
| public String getSubProtocol() { | ||
| return "amqp"; | ||
| } | ||
|
|
||
| @Override | ||
| public WireFormat getWireFormat() { | ||
| return frameReader.getWireFormat(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStop(ServiceStopper stopper) throws Exception { | ||
| // Currently nothing needed here since we have no async workers. | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStart() throws Exception { | ||
| if (outputSink == null) { | ||
| throw new IllegalStateException("Transport started before output sink assigned."); | ||
| } | ||
|
|
||
| // Currently nothing needed here since we have no async workers. | ||
| } | ||
|
|
||
| //----- WebSocket event hooks --------------------------------------------// | ||
|
|
||
| @Override | ||
| public void onWebSocketText(String data) throws IOException { | ||
| onException(new IOException("Illegal text content receive on AMQP WebSocket channel.")); | ||
| } | ||
|
|
||
| @Override | ||
| public void onWebSocketBinary(ByteBuffer data) throws IOException { | ||
| try { | ||
| frameReader.parse(data); | ||
| } catch (Exception e) { | ||
| throw IOExceptionSupport.create(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onWebSocketClosed() throws IOException { | ||
| onException(new IOException("Unexpected close of AMQP WebSocket channel.")); | ||
| } | ||
|
|
||
| //----- AMQP Frame Data event hook ---------------------------------------// | ||
|
|
||
| @Override | ||
| public void onFrame(Object frame) { | ||
| doConsume(frame); | ||
| } | ||
| } |
88 changes: 88 additions & 0 deletions
88
activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWSTransportFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.activemq.transport.amqp; | ||
|
|
||
| import java.io.IOException; | ||
| import java.net.MalformedURLException; | ||
| import java.net.URI; | ||
| import java.net.UnknownHostException; | ||
| import java.util.Map; | ||
|
|
||
| import org.apache.activemq.broker.BrokerService; | ||
| import org.apache.activemq.broker.BrokerServiceAware; | ||
| import org.apache.activemq.transport.Transport; | ||
| import org.apache.activemq.transport.TransportFactory; | ||
| import org.apache.activemq.transport.TransportServer; | ||
| import org.apache.activemq.util.IntrospectionSupport; | ||
| import org.apache.activemq.wireformat.WireFormat; | ||
|
|
||
| /** | ||
| * Factory for creating WebSocket aware AMQP Transports. | ||
| */ | ||
| public class AmqpWSTransportFactory extends TransportFactory implements BrokerServiceAware { | ||
|
|
||
| private BrokerService brokerService = null; | ||
|
|
||
| @Override | ||
| protected String getDefaultWireFormatType() { | ||
| return "amqp"; | ||
| } | ||
|
|
||
| @Override | ||
| public TransportServer doBind(URI location) throws IOException { | ||
| throw new IOException("doBind() method not implemented! No Server over WS implemented."); | ||
| } | ||
|
|
||
| @Override | ||
| @SuppressWarnings("rawtypes") | ||
| public Transport compositeConfigure(Transport transport, WireFormat format, Map options) { | ||
| AmqpTransportFilter amqpTransport = new AmqpTransportFilter(transport, format, brokerService); | ||
|
|
||
| Map<String, Object> wireFormatOptions = IntrospectionSupport.extractProperties(options, "wireFormat."); | ||
|
|
||
| IntrospectionSupport.setProperties(amqpTransport, options); | ||
| IntrospectionSupport.setProperties(amqpTransport.getWireFormat(), wireFormatOptions); | ||
|
|
||
| // Now wrap the filter with the monitor | ||
| transport = createInactivityMonitor(amqpTransport, format); | ||
| IntrospectionSupport.setProperties(transport, options); | ||
|
|
||
| return super.compositeConfigure(transport, format, options); | ||
| } | ||
|
|
||
| /** | ||
| * Factory method to create a new transport | ||
| * | ||
| * @throws IOException | ||
| * @throws UnknownHostException | ||
| */ | ||
| @Override | ||
| protected Transport createTransport(URI location, WireFormat wireFormat) throws MalformedURLException, UnknownHostException, IOException { | ||
| return new AmqpWSTransport(location, wireFormat); | ||
| } | ||
|
|
||
| @Override | ||
| public void setBrokerService(BrokerService brokerService) { | ||
| this.brokerService = brokerService; | ||
| } | ||
|
|
||
| protected Transport createInactivityMonitor(AmqpTransportFilter transport, WireFormat format) { | ||
| AmqpInactivityMonitor monitor = new AmqpInactivityMonitor(transport, format); | ||
| transport.setInactivityMonitor(monitor); | ||
| return monitor; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
17 changes: 17 additions & 0 deletions
17
activemq-amqp/src/main/resources/META-INF/services/org/apache/activemq/transport/amqp+ws
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| ## --------------------------------------------------------------------------- | ||
| ## Licensed to the Apache Software Foundation (ASF) under one or more | ||
| ## contributor license agreements. See the NOTICE file distributed with | ||
| ## this work for additional information regarding copyright ownership. | ||
| ## The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| ## (the "License"); you may not use this file except in compliance with | ||
| ## the License. You may obtain a copy of the License at | ||
| ## | ||
| ## http://www.apache.org/licenses/LICENSE-2.0 | ||
| ## | ||
| ## Unless required by applicable law or agreed to in writing, software | ||
| ## distributed under the License is distributed on an "AS IS" BASIS, | ||
| ## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| ## See the License for the specific language governing permissions and | ||
| ## limitations under the License. | ||
| ## --------------------------------------------------------------------------- | ||
| class=org.apache.activemq.transport.amqp.AmqpWSTransportFactory |
Oops, something went wrong.