private String aspec;
private String state;
private Set<ConnectListener> connls = new HashSet<ConnectListener>();
+ private Set<NotifyListener> notls = new HashSet<NotifyListener>();
private Exception error;
public interface ConnectListener {
public void connected() throws Exception;
public void error(Exception cause);
}
-
+
public Connection(String aspec) {
this.aspec = aspec;
state = "idle";
}
pending = new LinkedList<Command>();
Command ccmd = new Command(".connect");
- ccmd.addListener(new Command.Listener() {
+ ccmd.new Listener() {
public void done(Response resp) throws Exception {
try {
checkver(resp);
}
}
}
- });
+ };
pending.offer(ccmd);
reader = new Reader();
writer = new Writer();
throw(new RuntimeException("Cannot call synchronous method with dispatch thread!"));
}
- public void syncConnect() throws ConnectException, ClosedException, InterruptedException {
+ public void syncConnect() throws ConnectException, InterruptedException {
checkthread();
final boolean[] donep = new boolean[] {false};
final Exception[] errp = new Exception[] {null};
}
}
if(errp[0] != null)
- throw(new ClosedException(errp[0]));
+ throw(new ConnectException("DC connection has been closed", errp[0]));
}
public void expectVersion(int reqver) {
return(error);
}
+ public void addNotifyListener(NotifyListener l) {
+ synchronized(notls) {
+ notls.add(l);
+ }
+ }
+
+ public void removeNotifyListener(NotifyListener l) {
+ synchronized(notls) {
+ notls.remove(l);
+ }
+ }
+
public synchronized void addConnectListener(ConnectListener l) {
if((state != "idle") && (state != "connecting"))
throw(new IllegalStateException("Already connected"));
}
}
- private void qcmd(Command cmd) {
+ public void qcmd(Command... cmds) {
synchronized(queue) {
- queue.offer(cmd);
+ for(Command cmd : cmds)
+ queue.offer(cmd);
queue.notifyAll();
}
}
+ public void qcmd(String... tokens) {
+ qcmd(new Command(tokens));
+ }
+
+ public Response ecmd(Command cmd) throws ClosedException, InterruptedException {
+ checkthread();
+ final boolean[] donep = new boolean[] {false};
+ final Response[] resp = new Response[] {null};
+ final Exception[] errp = new Exception[] {null};
+ Object l = cmd.new Listener() {
+ public synchronized void done(Response rsp) {
+ resp[0] = rsp;
+ donep[0] = true;
+ notifyAll();
+ }
+
+ public synchronized void error(Exception e) {
+ errp[0] = e;
+ donep[0] = true;
+ notifyAll();
+ }
+ };
+ qcmd(cmd);
+ synchronized(l) {
+ while(!donep[0]) {
+ l.wait();
+ }
+ }
+ if(errp[0] != null)
+ throw(new ClosedException(errp[0]));
+ return(resp[0]);
+ }
+
+ public Response ecmd(String... tokens) throws ClosedException, InterruptedException {
+ return(ecmd(new Command(tokens)));
+ }
+
static private class StopCondition extends Error {
final boolean normal;
out.append(' ');
out.append(quote(s));
}
+ out.append("\r\n");
w.write(out.toString());
+ w.flush();
}
} catch(IOException e) {
throw(new StopCondition(e, false));
queue.notifyAll();
}
resp.cmd.done(resp);
+ } else {
+ synchronized(notls) {
+ for(NotifyListener l : notls) {
+ l.notified(resp);
+ }
+ }
}
}
code = Integer.parseInt(ct.toString());
ct.setLength(0);
state = "start";
- continue eat;
} else {
ct.append(c);
}