diff options
Diffstat (limited to 'java/common/src')
7 files changed, 163 insertions, 3 deletions
diff --git a/java/common/src/main/java/org/apache/qpidity/Channel.java b/java/common/src/main/java/org/apache/qpidity/Channel.java index 8cd07f002a..f20c65e467 100644 --- a/java/common/src/main/java/org/apache/qpidity/Channel.java +++ b/java/common/src/main/java/org/apache/qpidity/Channel.java @@ -205,7 +205,7 @@ class Channel extends Invoker implements Handler<Frame> method(m); } - protected void invoke(Method m, Handler<Struct> handler) + protected <T> Future<T> invoke(Method m, Class<T> cls) { throw new UnsupportedOperationException(); } diff --git a/java/common/src/main/java/org/apache/qpidity/Future.java b/java/common/src/main/java/org/apache/qpidity/Future.java new file mode 100644 index 0000000000..8902446e95 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpidity/Future.java @@ -0,0 +1,37 @@ +/* + * + * 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.qpidity; + + +/** + * Future + * + * @author Rafael H. Schloming + */ + +public interface Future<T> +{ + + T get(); + + boolean isDone(); + +} diff --git a/java/common/src/main/java/org/apache/qpidity/Result.java b/java/common/src/main/java/org/apache/qpidity/Result.java new file mode 100644 index 0000000000..7fe6c869a4 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpidity/Result.java @@ -0,0 +1,30 @@ +/* + * + * 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.qpidity; + + +/** + * Result + * + * @author Rafael H. Schloming + */ + +public abstract class Result extends Struct {} diff --git a/java/common/src/main/java/org/apache/qpidity/Session.java b/java/common/src/main/java/org/apache/qpidity/Session.java index c8b3c7c5bb..3b33fde2df 100644 --- a/java/common/src/main/java/org/apache/qpidity/Session.java +++ b/java/common/src/main/java/org/apache/qpidity/Session.java @@ -23,6 +23,8 @@ package org.apache.qpidity; import java.nio.ByteBuffer; import java.util.HashMap; import java.util.Map; + + /** * Session * @@ -217,9 +219,86 @@ public class Session extends Invoker } } - protected void invoke(Method m, Handler<Struct> handler) + private Map<Long,ResultFuture<?>> results = + new HashMap<Long,ResultFuture<?>>(); + + void result(long command, Struct result) + { + ResultFuture<?> future; + synchronized (results) + { + future = results.remove(command); + } + future.set(result); + } + + protected <T> Future<T> invoke(Method m, Class<T> klass) + { + long command = commandsOut; + ResultFuture<T> future = new ResultFuture<T>(klass); + synchronized (results) + { + results.put(command, future); + } + invoke(m); + return future; + } + + private class ResultFuture<T> implements Future<T> { - throw new UnsupportedOperationException(); + + private final Class<T> klass; + private T result; + + private ResultFuture(Class<T> klass) + { + this.klass = klass; + } + + private void set(Struct result) + { + synchronized (this) + { + this.result = klass.cast(result); + notifyAll(); + } + } + + public T get(long timeout, int nanos) + { + synchronized (this) + { + while (!isDone()) + { + try + { + wait(timeout, nanos); + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + } + } + + return result; + } + + public T get(long timeout) + { + return get(timeout, 0); + } + + public T get() + { + return get(0); + } + + public boolean isDone() + { + return result != null; + } + } } diff --git a/java/common/src/main/java/org/apache/qpidity/SessionDelegate.java b/java/common/src/main/java/org/apache/qpidity/SessionDelegate.java index 5d62a57e93..1287e3dc1a 100644 --- a/java/common/src/main/java/org/apache/qpidity/SessionDelegate.java +++ b/java/common/src/main/java/org/apache/qpidity/SessionDelegate.java @@ -34,6 +34,11 @@ public abstract class SessionDelegate extends Delegate<Session> public abstract void data(Session ssn, Frame frame); + @Override public void executionResult(Session ssn, ExecutionResult result) + { + ssn.result(result.getCommandId(), result.getData()); + } + @Override public void executionComplete(Session ssn, ExecutionComplete excmp) { RangeSet ranges = excmp.getRangedExecutionSet(); diff --git a/java/common/src/main/java/org/apache/qpidity/ToyBroker.java b/java/common/src/main/java/org/apache/qpidity/ToyBroker.java index 4a33122d37..651241f63c 100644 --- a/java/common/src/main/java/org/apache/qpidity/ToyBroker.java +++ b/java/common/src/main/java/org/apache/qpidity/ToyBroker.java @@ -60,6 +60,12 @@ class ToyBroker extends SessionDelegate System.out.println("declared queue: " + qd.getQueue()); } + @Override public void queueQuery(Session ssn, QueueQuery qq) + { + QueueQueryResult result = new QueueQueryResult().queue(qq.getQueue()); + ssn.executionResult(qq.getId(), result); + } + @Override public void messageTransfer(Session ssn, MessageTransfer xfr) { this.xfr = xfr; diff --git a/java/common/src/main/java/org/apache/qpidity/ToyClient.java b/java/common/src/main/java/org/apache/qpidity/ToyClient.java index 4ec838bc35..e325fb93be 100644 --- a/java/common/src/main/java/org/apache/qpidity/ToyClient.java +++ b/java/common/src/main/java/org/apache/qpidity/ToyClient.java @@ -85,6 +85,9 @@ class ToyClient extends SessionDelegate ssn.data("this should be rejected"); ssn.endData(); ssn.sync(); + + Future<QueueQueryResult> future = ssn.queueQuery("asdf"); + System.out.println(future.get().getQueue()); } } |