You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by de...@apache.org on 2010/09/16 14:15:54 UTC
svn commit: r997709 - in
/activemq/activemq-apollo/trunk/apollo-stomp/src/test: resources/
scala/org/apache/activemq/apollo/stomp/
scala/org/apache/activemq/apollo/stomp/perf/
Author: dejanb
Date: Thu Sep 16 12:15:54 2010
New Revision: 997709
URL: http://svn.apache.org/viewvc?rev=997709&view=rev
Log:
started work on stomp test
Added:
activemq/activemq-apollo/trunk/apollo-stomp/src/test/resources/
activemq/activemq-apollo/trunk/apollo-stomp/src/test/resources/activemq-stomp.xml
activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompClient.scala
activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompTest.scala
Modified:
activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompLoadClient.scala
Added: activemq/activemq-apollo/trunk/apollo-stomp/src/test/resources/activemq-stomp.xml
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/test/resources/activemq-stomp.xml?rev=997709&view=auto
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/test/resources/activemq-stomp.xml (added)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/test/resources/activemq-stomp.xml Thu Sep 16 12:15:54 2010
@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<!--
+ 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.
+-->
+<broker basedir="./activemq-data/default" rev="1" enabled="true" id="default" xmlns="http://activemq.apache.org/schema/activemq/apollo">
+ <notes>A default configuration</notes>
+ <virtual-host purge-on-startup="false" auto-create-queues="true" enabled="true" id="default">
+ <host-name>localhost</host-name>
+ </virtual-host>
+ <connectors advertise="tcp://localhost:61613" protocol="stomp" bind="tcp://0.0.0.0:61613" enabled="true" id="default"/>
+</broker>
\ No newline at end of file
Added: activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompClient.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompClient.scala?rev=997709&view=auto
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompClient.scala (added)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompClient.scala Thu Sep 16 12:15:54 2010
@@ -0,0 +1,107 @@
+/**
+ * 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.apollo.stomp
+
+import java.net.{Socket, InetSocketAddress}
+import org.apache.activemq.apollo.broker.ProtocolException
+import org.fusesource.hawtbuf.AsciiBuffer
+import _root_.org.fusesource.hawtbuf.{ByteArrayOutputStream => BAOS}
+import java.io._
+
+/**
+ * A simple Stomp client used for testing purposes
+ */
+ class StompClient {
+
+ var socket:Socket = new Socket
+ var out:OutputStream = null
+ var in:InputStream = null
+ val bufferSize = 64*1204
+
+ def open(host: String, port: Int) = {
+ socket = new Socket
+ socket.connect(new InetSocketAddress(host, port))
+ socket.setSoLinger(true, 0)
+ out = new BufferedOutputStream(socket.getOutputStream, bufferSize)
+ in = new BufferedInputStream(socket.getInputStream, bufferSize)
+ }
+
+ def close() = {
+ socket.close
+ }
+
+ def send(frame:String) = {
+ out.write(frame.getBytes("UTF-8"))
+ out.write(0)
+ out.write('\n')
+ out.flush
+ }
+
+ def send(frame:Array[Byte]) = {
+ out.write(frame)
+ out.write(0)
+ out.write('\n')
+ out.flush
+ }
+
+ def skip():Unit = {
+ var c = in.read
+ while( c >= 0 ) {
+ if( c==0 ) {
+ return
+ }
+ c = in.read()
+ }
+ throw new EOFException()
+ }
+
+ def receive():String = {
+ val buffer = new BAOS()
+ var c = in.read
+ while( c >= 0 ) {
+ if( c==0 ) {
+ return new String(buffer.toByteArray, "UTF-8")
+ }
+ buffer.write(c)
+ c = in.read()
+ }
+ throw new EOFException()
+ }
+
+ def receiveAscii():AsciiBuffer = {
+ val buffer = new BAOS()
+ var c = in.read
+ while( c >= 0 ) {
+ if( c==0 ) {
+ return buffer.toBuffer.ascii
+ }
+ buffer.write(c)
+ c = in.read()
+ }
+ throw new EOFException()
+ }
+
+ def receive(expect:String):String = {
+ val rc = receive()
+ if( !rc.startsWith(expect) ) {
+ throw new ProtocolException("Expected "+expect)
+ }
+ rc
+ }
+
+ }
\ No newline at end of file
Added: activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompTest.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompTest.scala?rev=997709&view=auto
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompTest.scala (added)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/StompTest.scala Thu Sep 16 12:15:54 2010
@@ -0,0 +1,50 @@
+/**
+ * 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.apollo.stomp
+
+import org.scalatest.matchers.ShouldMatchers
+import org.scalatest.{BeforeAndAfterAll, FunSuite}
+import org.apache.activemq.apollo.util.FunSuiteSupport
+import org.apache.activemq.apollo.broker.{Broker, BrokerFactory}
+
+class StompTest extends FunSuiteSupport with ShouldMatchers {
+
+ var broker:Broker = null
+
+
+ test("Stomp Connect") {
+ val client = new StompClient
+ client.open("localhost", 61613)
+ client.send("""CONNECT
+
+""")
+ val frame = client.receive()
+ frame should startWith ("CONNECTED")
+ }
+
+ override protected def beforeAll() = {
+ val uri = "xml:classpath:activemq-stomp.xml"
+ info("Loading broker configuration from the classpath with URI: " + uri)
+ broker = BrokerFactory.createBroker(uri, true)
+ Thread.sleep(1000); //TODO implement waitUntilStarted
+ }
+
+ override protected def afterAll() = {
+ broker.stop
+ }
+
+}
\ No newline at end of file
Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompLoadClient.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompLoadClient.scala?rev=997709&r1=997708&r2=997709&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompLoadClient.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompLoadClient.scala Thu Sep 16 12:15:54 2010
@@ -27,7 +27,7 @@ import java.net.{ProtocolException, Inet
import java.lang.String._
import java.util.concurrent.TimeUnit._
import collection.mutable.Map
-import org.apache.activemq.apollo.stomp.Stomp
+import org.apache.activemq.apollo.stomp.{StompClient, Stomp}
/**
*
@@ -176,88 +176,6 @@ object StompLoadClient {
def destination(i:Int) = "/"+destinationType+"/"+destinationName+"-"+(i%destinationCount)
-
- class StompClient {
-
- var socket:Socket = new Socket
- var out:OutputStream = null
- var in:InputStream = null
-
- def open(host: String, port: Int) = {
- socket = new Socket
- socket.connect(new InetSocketAddress(host, port))
- socket.setSoLinger(true, 0)
- out = new BufferedOutputStream(socket.getOutputStream, bufferSize)
- in = new BufferedInputStream(socket.getInputStream, bufferSize)
- }
-
- def close() = {
- socket.close
- }
-
- def flush() = {
- out.flush
- }
-
- def send(frame:String) = {
- out.write(frame.getBytes("UTF-8"))
- out.write(0)
- out.write('\n')
- }
-
- def send(frame:Array[Byte]) = {
- out.write(frame)
- out.write(0)
- out.write('\n')
- }
-
- def skip():Unit = {
- var c = in.read
- while( c >= 0 ) {
- if( c==0 ) {
- return
- }
- c = in.read()
- }
- throw new EOFException()
- }
-
- def receive():String = {
- val buffer = new ByteArrayOutputStream(messageSize+200)
- var c = in.read
- while( c >= 0 ) {
- if( c==0 ) {
- return new String(buffer.toByteArray, "UTF-8")
- }
- buffer.write(c)
- c = in.read()
- }
- throw new EOFException()
- }
-
- def receiveAscii():AsciiBuffer = {
- val buffer = new BAOS(messageSize+200)
- var c = in.read
- while( c >= 0 ) {
- if( c==0 ) {
- return buffer.toBuffer.ascii
- }
- buffer.write(c)
- c = in.read()
- }
- throw new EOFException()
- }
-
- def receive(expect:String):String = {
- val rc = receive()
- if( !rc.trimFront.startsWith(expect) ) {
- throw new ProtocolException("Expected "+expect)
- }
- rc
- }
-
- }
-
class ClientSupport extends Thread {
@@ -270,7 +188,6 @@ object StompLoadClient {
client.send("""CONNECT
""")
- client.flush
client.receive("CONNECTED")
proc
} catch {
@@ -320,12 +237,10 @@ object StompLoadClient {
client.send(content)
if( syncSend ) {
// waits for the reply..
- client.flush
client.skip
}
producerCounter.incrementAndGet()
if(producerSleep > 0) {
- client.flush
Thread.sleep(producerSleep)
}
i += 1
@@ -364,7 +279,6 @@ object StompLoadClient {
"destination:"+destination(id)+"\n"+
"\n")
- client.flush
receiveLoop
}
}
@@ -384,7 +298,6 @@ ACK
message-id:"""+msgId+"""
""")
- client.flush
} else {
client.skip