You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2008/08/13 12:39:12 UTC
svn commit: r685506 - in
/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport:
Echo.java Sink.java
Author: rhs
Date: Wed Aug 13 03:39:10 2008
New Revision: 685506
URL: http://svn.apache.org/viewvc?rev=685506&view=rev
Log:
removed dead code from message Echo utility, and added a message Sink utility
Added:
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java
Modified:
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Echo.java
Modified: incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Echo.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Echo.java?rev=685506&r1=685505&r2=685506&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Echo.java (original)
+++ incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Echo.java Wed Aug 13 03:39:10 2008
@@ -35,11 +35,8 @@
public class Echo extends SessionDelegate
{
- private MessageTransfer xfr = null;
-
public void messageTransfer(Session ssn, MessageTransfer xfr)
{
- this.xfr = xfr;
ssn.invoke(xfr);
ssn.processed(xfr);
}
Added: incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java?rev=685506&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java (added)
+++ incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java Wed Aug 13 03:39:10 2008
@@ -0,0 +1,137 @@
+/*
+ *
+ * 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.qpid.transport;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.qpid.transport.network.ConnectionBinding;
+import org.apache.qpid.transport.network.io.IoAcceptor;
+
+/**
+ * Sink
+ *
+ */
+
+public class Sink extends SessionDelegate
+{
+
+ private static final String FORMAT_HDR = "%-12s %-18s %-18s %-18s";
+ private static final String FORMAT_ROW = "SSN#%-8X %-18s %-18s %-18s";
+
+ private long interval = 100000;
+ private long start = System.currentTimeMillis();
+ private long count = 0;
+ private long bytes = 0;
+ private long interval_start = start;
+ private long bytes_start = bytes;
+ private long time = start;
+ private int id = System.identityHashCode(this);
+
+ public Sink()
+ {
+ }
+
+ private double msg_rate()
+ {
+ return 1000 * (double) count / (double) (time - start);
+ }
+
+ private double byte_rate()
+ {
+ return (1000 * (double) bytes / (double) (time - start)) / (1024*1024);
+ }
+
+ private double msg_interval_rate()
+ {
+ return 1000 * (double) interval / (double) (time - interval_start);
+ }
+
+ private double byte_interval_rate()
+ {
+ return (1000 * (double) (bytes - bytes_start) / (double) (time - interval_start)) / (1024*1024);
+ }
+
+ private String rates()
+ {
+ return String.format("%.2f/%.2f", msg_rate(), byte_rate());
+ }
+
+ private String interval_rates()
+ {
+ return String.format("%.2f/%.2f", msg_interval_rate(), byte_interval_rate());
+ }
+
+ private String counts()
+ {
+ return String.format("%d/%.2f", count, ((double) bytes)/(1024*1024));
+ }
+
+ public void messageTransfer(Session ssn, MessageTransfer xfr)
+ {
+ count++;
+ bytes += xfr.getBody().remaining();
+ if ((count % interval) == 0)
+ {
+ time = System.currentTimeMillis();
+ System.out.println
+ (String.format
+ (FORMAT_ROW, id, counts(), rates(), interval_rates()));
+ interval_start = time;
+ bytes_start = bytes;
+ }
+ ssn.processed(xfr);
+ }
+
+ public static final void main(String[] args) throws IOException
+ {
+ ConnectionDelegate delegate = new ConnectionDelegate()
+ {
+
+ public SessionDelegate getSessionDelegate()
+ {
+ return new Sink();
+ }
+
+ public void exception(Throwable t)
+ {
+ t.printStackTrace();
+ }
+
+ public void closed() {}
+ };
+
+ //hack
+ delegate.setUsername("guest");
+ delegate.setPassword("guest");
+
+ IoAcceptor ioa = new IoAcceptor
+ ("0.0.0.0", 5672, new ConnectionBinding(delegate));
+ System.out.println
+ (String.format
+ (FORMAT_HDR, "Session", "Count/MBytes", "Cumulative Rate", "Interval Rate"));
+ System.out.println
+ (String.format
+ (FORMAT_HDR, "-------", "------------", "---------------", "-------------"));
+ ioa.start();
+ }
+
+}