activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jstrac...@apache.org
Subject svn commit: r411881 - in /incubator/activemq/trunk/activemq-core/src: main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
Date Mon, 05 Jun 2006 18:19:02 GMT
Author: jstrachan
Date: Mon Jun  5 11:19:01 2006
New Revision: 411881

URL: http://svn.apache.org/viewvc?rev=411881&view=rev
Log:
added test case and fix for AMQ-724

Added:
    incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
  (with props)
Modified:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java?rev=411881&r1=411880&r2=411881&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java
Mon Jun  5 11:19:01 2006
@@ -16,32 +16,17 @@
  */
 package org.apache.activemq.transport.stomp;
 
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-
-import org.apache.activemq.transport.TransportFactory;
-import org.apache.activemq.transport.TransportServer;
-import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
 
 /**
  * A <a href="http://stomp.codehaus.org/">Stomp</a> transport factory
  * 
  * @version $Revision: 1.1.1.1 $
  */
-public class StompTransportFactory extends TransportFactory {
+public class StompTransportFactory extends TcpTransportFactory {
 
-    public TransportServer doBind(String brokerId, URI location) throws IOException {
-        try {
-            URI tcpURI = new URI(
-                    "tcp://"+location.getHost()+
-                    (location.getPort()>=0 ? ":"+location.getPort() : "")+
-                    "?wireFormat=stomp"
-                    );
-            return TransportFactory.bind(brokerId, tcpURI);
-        } catch (URISyntaxException e) {
-            throw IOExceptionSupport.create(e);
-        }
+    protected String getDefaultWireFormatType() {
+        return "stomp";
     }
 
 }

Added: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java?rev=411881&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
(added)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
Mon Jun  5 11:19:01 2006
@@ -0,0 +1,229 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.stomp;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+
+import javax.jms.Connection;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import junit.framework.TestCase;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.Broker;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.transport.stomp.Stomp;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * 
+ * @version $Revision$
+ */
+public class StompSubscriptionRemoveTest extends TestCase {
+    private static final Log log = LogFactory.getLog(StompSubscriptionRemoveTest.class);
+
+    private Socket stompSocket;
+    private ByteArrayOutputStream inputBuffer;
+
+    /**
+     * @param args
+     * @throws Exception
+     */
+    public void testRemoveSubscriber() throws Exception {
+        BrokerService broker = new BrokerService();
+        broker.setPersistent(true);
+
+        broker.addConnector("stomp://localhost:61613").setName("Stomp");
+        broker.addConnector("tcp://localhost:61616").setName("Default");
+        broker.start();
+
+        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+        Connection connection = factory.createConnection();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer producer = session.createProducer(new ActiveMQQueue(getDestinationName()));
+        Message message = session.createTextMessage("Testas");
+        for (int idx = 0; idx < 2000; ++idx) {
+            producer.send(message);
+            log.debug("Sending: " + idx);
+        }
+        producer.close();
+        // consumer.close();
+        session.close();
+        connection.close();
+        broker.stop();
+        while (broker.isStarted()) {
+            Thread.sleep(1000);
+        }
+
+        broker = new BrokerService();
+        broker.setPersistent(true);
+
+        broker.addConnector("stomp://localhost:61613").setName("Stomp");
+        broker.addConnector("tcp://localhost:61616").setName("Default");
+        broker.start();
+
+        stompSocket = new Socket("localhost", 61613);
+        inputBuffer = new ByteArrayOutputStream();
+
+        String connect_frame = "CONNECT\n" + "login: brianm\n" + "passcode: wombats\n" +
"\n" + Stomp.NULL;
+        sendFrame(connect_frame);
+
+        String f = receiveFrame(100000);
+        String frame = "SUBSCRIBE\n" + "destination:/queue/" + getDestinationName() + "\n"
+ "ack:client\n\n" + Stomp.NULL;
+        sendFrame(frame);
+        int messagesCount = 0;
+        int count = 0;
+        while (count < 2) {
+            String receiveFrame = receiveFrame(10000);
+            DataInput input = new DataInputStream(new ByteArrayInputStream(receiveFrame.getBytes()));
+            String line;
+            while (true) {
+                line = input.readLine();
+                if (line == null) {
+                    throw new IOException("connection was closed");
+                }
+                else {
+                    line = line.trim();
+                    if (line.length() > 0) {
+                        break;
+                    }
+                }
+            }
+            line = input.readLine();
+            if (line == null) {
+                throw new IOException("connection was closed");
+            }
+            String messageId = line.substring(line.indexOf(':') + 1);
+            messageId = messageId.trim();
+            String ackmessage = "ACK\n" + "message-id:" + messageId + "\n\n" + Stomp.NULL;
+            sendFrame(ackmessage);
+            log.debug(receiveFrame);
+            //Thread.sleep(1000);
+            ++messagesCount;
+            ++count;
+        }
+        stompSocket.close();
+        Thread.sleep(10000);
+
+        // for (int idx = 0; idx < 500; ++idx) {
+        // producer.send(message);
+        // log.debug("Sending: " +idx);
+        // }
+
+        stompSocket = new Socket("localhost", 61613);
+        inputBuffer = new ByteArrayOutputStream();
+
+        connect_frame = "CONNECT\n" + "login: brianm\n" + "passcode: wombats\n" + "\n" +
Stomp.NULL;
+        sendFrame(connect_frame);
+
+        f = receiveFrame(100000);
+        frame = "SUBSCRIBE\n" + "destination:/queue/" + getDestinationName() + "\n" + "ack:client\n\n"
+ Stomp.NULL;
+        sendFrame(frame);
+        try {
+            while (count != 2000) {
+                String receiveFrame = receiveFrame(10000);
+                DataInput input = new DataInputStream(new ByteArrayInputStream(receiveFrame.getBytes()));
+                String line;
+                while (true) {
+                    line = input.readLine();
+                    if (line == null) {
+                        throw new IOException("connection was closed");
+                    }
+                    else {
+                        line = line.trim();
+                        if (line.length() > 0) {
+                            break;
+                        }
+                    }
+                }
+                line = input.readLine();
+                if (line == null) {
+                    throw new IOException("connection was closed");
+                }
+                String messageId = line.substring(line.indexOf(':') + 1);
+                messageId = messageId.trim();
+                String ackmessage = "ACK\n" + "message-id:" + messageId + "\n\n" + Stomp.NULL;
+                sendFrame(ackmessage);
+                log.debug("Received: " + receiveFrame);
+                //Thread.sleep(1000);
+                ++messagesCount;
+                ++count;
+            }
+
+        }
+        catch (IOException ex) {
+            // timeout
+        }
+        stompSocket.close();
+        broker.stop();
+        log.info("Total messages receved: " + messagesCount);
+        assertTrue("Messages receved after connection loss: " + messagesCount, messagesCount
>= 2000);
+
+        // The first ack messages has no chance complete, so we receiving more messages
+
+        // Don't know how to list subscriptions for the broker. Currently you
+        // can check using jmx console. You'll see
+        // Subscription whithout any connections
+    }
+
+    public void sendFrame(String data) throws Exception {
+        byte[] bytes = data.getBytes("UTF-8");
+        OutputStream outputStream = stompSocket.getOutputStream();
+        for (int i = 0; i < bytes.length; i++) {
+            outputStream.write(bytes[i]);
+        }
+        outputStream.flush();
+    }
+
+    public String receiveFrame(long timeOut) throws Exception {
+        stompSocket.setSoTimeout((int) timeOut);
+        InputStream is = stompSocket.getInputStream();
+        int c = 0;
+        for (;;) {
+            c = is.read();
+            if (c < 0) {
+                throw new IOException("socket closed.");
+            }
+            else if (c == 0) {
+                c = is.read();
+                byte[] ba = inputBuffer.toByteArray();
+                inputBuffer.reset();
+                return new String(ba, "UTF-8");
+            }
+            else {
+                inputBuffer.write(c);
+            }
+        }
+    }
+
+    protected String getDestinationName() {
+        return getClass().getName() + "." + getName();
+    }
+}

Propchange: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
------------------------------------------------------------------------------
    svn:keywords = Date Author Id Revision HeadURL

Propchange: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSubscriptionRemoveTest.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain



Mime
View raw message