activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r1393790 - in /activemq/trunk/activemq-amqp/src: main/java/org/apache/activemq/transport/amqp/ main/java/org/apache/activemq/transport/amqp/transform/ test/java/org/apache/activemq/transport/amqp/
Date Wed, 03 Oct 2012 22:01:08 GMT
Author: chirino
Date: Wed Oct  3 22:01:08 2012
New Revision: 1393790

URL: http://svn.apache.org/viewvc?rev=1393790&view=rev
Log:
AMQP impl: A simple send and receive is now working

Modified:
    activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
    activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/transform/AMQPNativeInboundTransformer.java
    activemq/trunk/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SwiftMQClientTest.java

Modified: activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java?rev=1393790&r1=1393789&r2=1393790&view=diff
==============================================================================
--- activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
(original)
+++ activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
Wed Oct  3 22:01:08 2012
@@ -86,7 +86,7 @@ class AmqpProtocolConverter {
                 int count = protonTransport.output(data, 0, size);
                 if (count > 0) {
                     final Buffer buffer = new Buffer(data, 0, count);
-                    System.out.println("writing: " + buffer.toString().substring(5).replaceAll("(..)",
"$1 "));
+//                    System.out.println("writing: " + buffer.toString().substring(5).replaceAll("(..)",
"$1 "));
                     amqpTransport.sendToAmqp(buffer);
                 } else {
                     done = true;
@@ -116,7 +116,7 @@ class AmqpProtocolConverter {
 
 
         try {
-            System.out.println("reading: " + frame.toString().substring(5).replaceAll("(..)",
"$1 "));
+//            System.out.println("reading: " + frame.toString().substring(5).replaceAll("(..)",
"$1 "));
             protonTransport.input(frame.data, frame.offset, frame.length);
         } catch (Throwable e) {
             handleException(new AmqpProtocolException("Could not decode AMQP frame: " + frame,
true, e));

Modified: activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/transform/AMQPNativeInboundTransformer.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/transform/AMQPNativeInboundTransformer.java?rev=1393790&r1=1393789&r2=1393790&view=diff
==============================================================================
--- activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/transform/AMQPNativeInboundTransformer.java
(original)
+++ activemq/trunk/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/transform/AMQPNativeInboundTransformer.java
Wed Oct  3 22:01:08 2012
@@ -45,7 +45,7 @@ public class AMQPNativeInboundTransforme
         }
 
         rc.setLongProperty(prefixVendor + "MESSAGE_FORMAT", messageFormat);
-        rc.setBooleanProperty(prefixVendor + "NATIVE", false);
+        rc.setBooleanProperty(prefixVendor + "NATIVE", true);
         return rc;
     }
 }

Modified: activemq/trunk/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SwiftMQClientTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SwiftMQClientTest.java?rev=1393790&r1=1393789&r2=1393790&view=diff
==============================================================================
--- activemq/trunk/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SwiftMQClientTest.java
(original)
+++ activemq/trunk/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SwiftMQClientTest.java
Wed Oct  3 22:01:08 2012
@@ -23,6 +23,7 @@ import com.swiftmq.amqp.v100.messaging.A
 import com.swiftmq.amqp.v100.types.AMQPString;
 import com.swiftmq.amqp.v100.types.AMQPType;
 import org.junit.Test;
+import static org.junit.Assert.*;
 
 /**
  * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
@@ -50,39 +51,38 @@ public class SwiftMQClientTest extends A
             });
             connection.connect();
             {
-                String data = String.format("%010d", 0);
 
                 Session session = connection.createSession(10, 10);
                 Producer p = session.createProducer(queue, qos);
                 for (int i = 0; i < nMsgs; i++) {
                     AMQPMessage msg = new AMQPMessage();
-                    String s = "Message #" + (i + 1);
-                    System.out.println("Sending " + s);
-                    msg.setAmqpValue(new AmqpValue(new AMQPString(s + ", data: " + data)));
+                    System.out.println("Sending " + i);
+                    msg.setAmqpValue(new AmqpValue(new AMQPString(String.format("%010d",
i))));
                     p.send(msg);
                 }
                 p.close();
                 session.close();
             }
 
-//            {
-//                Session session = connection.createSession(10, 10);
-//                Consumer c = session.createConsumer(queue, 100, qos, true, null);
-//
-//                // Receive messages non-transacted
-//                for (int i = 0; i < nMsgs; i++) {
-//                    AMQPMessage msg = c.receive();
-//                    final AMQPType value = msg.getAmqpValue().getValue();
-//                    if (value instanceof AMQPString) {
-//                        AMQPString s = (AMQPString) value;
-//                        System.out.println("Received: " + s.getValue());
-//                    }
-//                    if (!msg.isSettled())
-//                        msg.accept();
-//                }
-//                c.close();
-//                session.close();
-//            }
+            {
+                Session session = connection.createSession(10, 10);
+                Consumer c = session.createConsumer(queue, 100, qos, true, null);
+
+                // Receive messages non-transacted
+                for (int i = 0; i < nMsgs; i++) {
+                    AMQPMessage msg = c.receive();
+                    final AMQPType value = msg.getAmqpValue().getValue();
+                    if (value instanceof AMQPString) {
+                        String s = ((AMQPString) value).getValue();
+                        assertEquals(String.format("%010d", i), s);
+                        System.out.println("Received: " + i);
+                    }
+                    if (!msg.isSettled())
+                        msg.accept();
+                }
+                c.close();
+                session.close();
+            }
             connection.close();
         } catch (Exception e) {
             e.printStackTrace();



Mime
View raw message