Make ServerApi flushable, and flush in ping-pong test
This commit is contained in:
parent
8b3f23e600
commit
76cf11c69a
|
@ -4,12 +4,13 @@
|
||||||
|
|
||||||
package hop;
|
package hop;
|
||||||
|
|
||||||
|
import java.io.Flushable;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.UUID;
|
import java.util.UUID;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
*/
|
*/
|
||||||
public class ServerApi {
|
public class ServerApi implements Flushable {
|
||||||
public NodeContainer _container;
|
public NodeContainer _container;
|
||||||
public String _serverName;
|
public String _serverName;
|
||||||
public String _kName;
|
public String _kName;
|
||||||
|
@ -37,9 +38,13 @@ public class ServerApi {
|
||||||
_container.post(_serverName, sink, message, null);
|
_container.post(_serverName, sink, message, null);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void flush() throws IOException {
|
||||||
|
_container.flush(_serverName);
|
||||||
|
}
|
||||||
|
|
||||||
public synchronized Object subscribe(String source, Object filter, String sink, String name) throws InterruptedException, IOException {
|
public synchronized Object subscribe(String source, Object filter, String sink, String name) throws InterruptedException, IOException {
|
||||||
send(source, SexpMessage.subscribe(filter, sink, name, _container.getName(), _kName));
|
send(source, SexpMessage.subscribe(filter, sink, name, _container.getName(), _kName));
|
||||||
_container.flush(_serverName);
|
flush();
|
||||||
SexpList reply = _nextReply();
|
SexpList reply = _nextReply();
|
||||||
assert reply.getBytes(0).getDataString().equals(SexpMessage._subscribe_ok);
|
assert reply.getBytes(0).getDataString().equals(SexpMessage._subscribe_ok);
|
||||||
return reply.get(1);
|
return reply.get(1);
|
||||||
|
@ -55,13 +60,13 @@ public class ServerApi {
|
||||||
|
|
||||||
public void unsubscribe(String source, Object token) throws IOException {
|
public void unsubscribe(String source, Object token) throws IOException {
|
||||||
send(source, SexpMessage.unsubscribe(token));
|
send(source, SexpMessage.unsubscribe(token));
|
||||||
_container.flush(_serverName);
|
flush();
|
||||||
/* TODO: optional synchronous reply? */
|
/* TODO: optional synchronous reply? */
|
||||||
}
|
}
|
||||||
|
|
||||||
public synchronized Object create(String nodeClassName, Object arg) throws InterruptedException, IOException {
|
public synchronized Object create(String nodeClassName, Object arg) throws InterruptedException, IOException {
|
||||||
send("factory", SexpMessage.create(nodeClassName, arg, _container.getName(), _kName));
|
send("factory", SexpMessage.create(nodeClassName, arg, _container.getName(), _kName));
|
||||||
_container.flush(_serverName);
|
flush();
|
||||||
SexpList reply = _nextReply();
|
SexpList reply = _nextReply();
|
||||||
SexpBytes selector = reply.getBytes(0);
|
SexpBytes selector = reply.getBytes(0);
|
||||||
if (selector.equals(SexpMessage._create_ok)) return null;
|
if (selector.equals(SexpMessage._create_ok)) return null;
|
||||||
|
|
|
@ -21,6 +21,7 @@ public class Test3 {
|
||||||
NodeContainer nc = new NodeContainer();
|
NodeContainer nc = new NodeContainer();
|
||||||
|
|
||||||
System.out.println("Hostname: " + hostname);
|
System.out.println("Hostname: " + hostname);
|
||||||
|
|
||||||
System.out.println("Container: " + nc.getName());
|
System.out.println("Container: " + nc.getName());
|
||||||
|
|
||||||
Relay r = new Relay(nc, hostname);
|
Relay r = new Relay(nc, hostname);
|
||||||
|
@ -33,7 +34,7 @@ public class Test3 {
|
||||||
api.post("q1", null, Integer.toString(i), null);
|
api.post("q1", null, Integer.toString(i), null);
|
||||||
count++;
|
count++;
|
||||||
if ((count % 100000) == 0) {
|
if ((count % 100000) == 0) {
|
||||||
r.flush();
|
api.flush();
|
||||||
long now = System.currentTimeMillis();
|
long now = System.currentTimeMillis();
|
||||||
double delta = (now - startTime) / 1000.0;
|
double delta = (now - startTime) / 1000.0;
|
||||||
System.out.println("Sent "+count+" messages in "+delta+" seconds, rate = " + (count / delta) + " Hz");
|
System.out.println("Sent "+count+" messages in "+delta+" seconds, rate = " + (count / delta) + " Hz");
|
||||||
|
|
|
@ -39,6 +39,7 @@ public class TestPingPong {
|
||||||
Object x = sub.getQueue().take();
|
Object x = sub.getQueue().take();
|
||||||
//System.out.println("Message: " + x);
|
//System.out.println("Message: " + x);
|
||||||
api.post("rep", "reply", SexpList.with("ok").and(x), null);
|
api.post("rep", "reply", SexpList.with("ok").and(x), null);
|
||||||
|
api.flush();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -57,7 +58,8 @@ public class TestPingPong {
|
||||||
long startTime = System.currentTimeMillis();
|
long startTime = System.currentTimeMillis();
|
||||||
for (int i = 0; i < 100000; i++) {
|
for (int i = 0; i < 100000; i++) {
|
||||||
api.post("req", "request", Integer.toString(i), null);
|
api.post("req", "request", Integer.toString(i), null);
|
||||||
sub.getQueue().take();
|
api.flush();
|
||||||
|
sub.getQueue().take();
|
||||||
int j = i + 1;
|
int j = i + 1;
|
||||||
if ((j % 100) == 0) {
|
if ((j % 100) == 0) {
|
||||||
long now = System.currentTimeMillis();
|
long now = System.currentTimeMillis();
|
||||||
|
|
Loading…
Reference in New Issue