6 import java.util.function.*;
7 import java.util.concurrent.*;
8 import java.util.logging.*;
11 import java.nio.channels.*;
13 public class EventServer implements Runnable {
14 private static final double timeout = 5;
15 private static final Logger log = Logger.getLogger("jagi.server");
16 private final ServerSocketChannel sk;
17 private final Function handler;
18 private final Driver ev = Driver.get();
19 private final ExecutorService handlers = new ThreadPoolExecutor(0, Runtime.getRuntime().availableProcessors() * 2,
20 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(64),
21 tgt -> new Thread(tgt, "Request handler thread"));
23 public EventServer(ServerSocketChannel sk, Function handler) {
25 sk.configureBlocking(false);
26 } catch(IOException e) {
27 throw(new RuntimeException(e));
30 this.handler = handler;
33 public static class Request {
34 public final Map<Object, Object> env;
35 public final SocketChannel sk;
37 public Request(Map<Object, Object> env, SocketChannel sk) {
43 ArrayList<Object> cleanup = new ArrayList<>((Collection<?>)env.get("jagi.cleanup"));
45 RuntimeException ce = null;
46 for(Object obj : cleanup) {
47 if(obj instanceof AutoCloseable) {
49 ((AutoCloseable)obj).close();
50 } catch(Exception e) {
52 ce = new RuntimeException("error(s) occurred during cleanup");
62 protected void error(Request req, Throwable error) {
63 log.log(Level.WARNING, "uncaught exception while handling request", error);
66 public static abstract class ChainWatcher implements Watcher {
67 private Runnable then;
68 public ChainWatcher then(Runnable then) {this.then = then; return(this);}
76 public static class BufferedOutput extends ChainWatcher {
77 public final SocketChannel sk;
78 public final ByteBuffer buf;
79 private double lastwrite;
81 public BufferedOutput(SocketChannel sk, ByteBuffer buf) {
86 public void added(Driver d) {lastwrite = d.time();}
87 public SelectableChannel channel() {return(sk);}
88 public int events() {return((buf.remaining() > 0) ? SelectionKey.OP_WRITE : -1);}
89 public double timeout() {return(lastwrite + timeout);}
91 public void handle(int events) throws IOException {
92 double now = Driver.current().time();
93 if((events & SelectionKey.OP_WRITE) != 0) {
97 if(now > lastwrite + timeout)
98 buf.position(buf.limit());
102 public static class TransferOutput extends ChainWatcher {
103 public final SocketChannel sk;
104 public final ReadableByteChannel in;
105 private final ByteBuffer buf;
106 private boolean eof = false;
107 private double lastwrite;
109 public TransferOutput(SocketChannel sk, ReadableByteChannel in) {
112 buf = ByteBuffer.allocate(65536);
116 public void added(Driver d) {lastwrite = d.time();}
117 public SelectableChannel channel() {return(sk);}
118 public int events() {return((eof && (buf.remaining() == 0)) ? -1 : SelectionKey.OP_WRITE);}
119 public double timeout() {return(lastwrite + timeout);}
121 public void handle(int events) throws IOException {
122 if(!eof && (buf.remaining() == 0)) {
124 while(buf.remaining() > 0) {
129 double now = Driver.current().time();
130 if((events & SelectionKey.OP_WRITE) != 0) {
131 if(sk.write(buf) > 0)
134 if(now > lastwrite + timeout) {
136 buf.position(buf.limit());
140 public void close() {
143 } catch(IOException e) {
144 log.log(Level.WARNING, "failed to close transfer channel: " + in, e);
151 public static class TransferInput extends ChainWatcher {
152 public final SocketChannel sk;
153 public final WritableByteChannel out;
154 private final ByteBuffer buf;
155 private boolean eof = false;
156 private double lastread;
158 public TransferInput(SocketChannel sk, WritableByteChannel out) {
161 buf = ByteBuffer.allocate(65536);
165 public void added(Driver d) {lastread = d.time();}
166 public SelectableChannel channel() {return(sk);}
167 public int events() {return(eof ? -1 : SelectionKey.OP_READ);}
168 public double timeout() {return(lastread + timeout);}
170 public void handle(int events) throws IOException {
171 double now = Driver.current().time();
172 if((events & SelectionKey.OP_READ) != 0) {
174 int rv = sk.read(buf);
180 while(buf.remaining() > 0)
183 if(now > lastread + timeout) {
185 buf.position(buf.limit());
189 public void close() {
192 } catch(IOException e) {
193 log.log(Level.WARNING, "failed to close transfer channel: " + out, e);
200 protected void respond(Request req, String status, Map resp) {
201 Object output = resp.get("jagi.output");
202 ByteArrayOutputStream buf = new ByteArrayOutputStream();
204 Writer head = new OutputStreamWriter(buf, Utils.UTF8);
205 head.write("Status: ");
208 for(Iterator it = resp.entrySet().iterator(); it.hasNext();) {
209 Map.Entry ent = (Map.Entry)it.next();
210 Object val = ent.getValue();
211 if((ent.getKey() instanceof String) && (val != null)) {
212 String key = (String)ent.getKey();
213 if(key.startsWith("http.")) {
214 String nm = key.substring(5);
215 if(nm.equalsIgnoreCase("status"))
217 if(val instanceof Collection) {
218 for(Object part : (Collection)val) {
221 head.write(part.toString());
227 head.write(val.toString());
235 } catch(IOException e) {
236 throw(new RuntimeException("cannot happen"));
240 out = new BufferedOutput(req.sk, ByteBuffer.allocate(0));
241 } else if(output instanceof byte[]) {
242 out = new BufferedOutput(req.sk, ByteBuffer.wrap((byte[])output));
243 } else if(output instanceof ByteBuffer) {
244 out = new BufferedOutput(req.sk, (ByteBuffer)output);
245 } else if(output instanceof String) {
246 out = new BufferedOutput(req.sk, ByteBuffer.wrap(((String)output).getBytes(Utils.UTF8)));
247 } else if(output instanceof CharSequence) {
248 out = new BufferedOutput(req.sk, Utils.UTF8.encode(CharBuffer.wrap((CharSequence)output)));
249 } else if(output instanceof InputStream) {
250 out = new TransferOutput(req.sk, Channels.newChannel((InputStream)output));
251 } else if(output instanceof ReadableByteChannel) {
252 out = new TransferOutput(req.sk, (ReadableByteChannel)output);
254 throw(new IllegalArgumentException("response-body: " + output));
256 out.then(() -> submit(req::close));
257 ev.add(new BufferedOutput(req.sk, ByteBuffer.wrap(buf.toByteArray())).then(() -> ev.add(out)));
260 @SuppressWarnings("unchecked")
261 protected void handle(Request req, Function handler) {
262 boolean handoff = false;
264 Throwable error = null;
266 Map resp = (Map)handler.apply(req.env);
268 if((st = (String)resp.get("jagi.status")) != null) {
269 Function next = (Function)resp.get("jagi.next");
272 Object sink = resp.get("jagi.input-sink");
273 if(sink instanceof WritableByteChannel) {
274 ev.add(new TransferInput(req.sk, (WritableByteChannel)sink).then(() -> submit(() -> handle(req, next))));
275 } else if(sink instanceof OutputStream) {
276 ev.add(new TransferInput(req.sk, Channels.newChannel((OutputStream)sink)).then(() -> submit(() -> handle(req, next))));
278 throw(new IllegalArgumentException("input-sink: " + sink));
283 throw(new IllegalArgumentException("jagi.status: " + st));
285 } else if((st = (String)resp.get("http.status")) != null) {
286 respond(req, st, resp);
289 } catch(Throwable t) {
296 } catch(Throwable ce) {
300 error.addSuppressed(ce);
305 } catch(Throwable t) {
310 protected void submit(Runnable task) {
311 handlers.submit(task);
314 class Client implements Watcher {
315 final SocketChannel sk;
317 boolean eof = false, handoff = false;
319 ByteBuffer head = null;
320 Map<Object, Object> env = null;
322 Client(SocketChannel sk) {
326 public void added(Driver d) {lastread = d.time();}
327 public SelectableChannel channel() {return(sk);}
328 public double timeout() {return(lastread + timeout);}
329 public int events() {
333 return(SelectionKey.OP_READ);
337 boolean readhead() throws IOException {
339 ByteBuffer buf = ByteBuffer.allocate(1);
342 int rv = sk.read(buf);
349 lastread = Driver.current().time();
351 if((c >= '0') && (c <= '9')) {
352 headlen = (headlen * 10) + (c - '0');
353 } else if(c == ':') {
354 head = ByteBuffer.allocate(headlen + 1);
364 if(head.remaining() == 0) {
365 if(head.get(head.limit() - 1) != ',') {
366 /* Unterminated netstring */
370 head.limit(head.limit() - 1);
371 env = Jagi.mkenv(Scgi.splithead(head), sk);
374 int rv = sk.read(head);
384 public void handle(int events) throws IOException {
385 if((events & SelectionKey.OP_READ) != 0) {
386 if((env == null) && !readhead())
388 Request req = new Request(env, sk);
389 submit(() -> EventServer.this.handle(req, handler));
392 if(Driver.current().time() > (lastread + timeout))
396 public void close() {
400 } catch(IOException e) {
406 class Accepter implements Watcher {
407 boolean closed = false;
409 public SelectableChannel channel() {return(sk);}
410 public int events() {return(SelectionKey.OP_ACCEPT);}
412 public void handle(int events) throws IOException {
413 if((events & SelectionKey.OP_ACCEPT) != 0) {
414 SocketChannel cl = sk.accept();
415 cl.configureBlocking(false);
416 Driver.current().add(new Client(cl));
420 public void close() {
429 Accepter main = new Accepter();
433 while(!main.closed) {
437 } catch(InterruptedException e) {
442 } catch(IOException e) {
443 throw(new RuntimeException(e));