Mercurial > projects > dreactor
changeset 11:5836613d16ac
reorg! reorg!
| author | rick@minifunk |
|---|---|
| date | Tue, 12 Aug 2008 16:59:56 -0400 |
| parents | e75a2e506b1d |
| children | d6a3cfe7c3de |
| files | asyncdreactor/core/AsyncVat.d asyncdreactor/core/Dispatcher.d asyncdreactor/protocol/Http11.d asyncdreactor/protocol/Http11Parser.d asyncdreactor/protocol/RawTcp.d asyncdreactor/protocol/RawUdp.d asyncdreactor/protocol/http11_parser.rl asyncdreactor/transport/AsyncSocketConduit.d asyncdreactor/util/ThreadSafeQueue.d dreactor/core/Dispatcher.d dreactor/core/Task.d dreactor/core/Vat.d dreactor/protocol/Protocol.d dreactor/protocol/RawTcp.d dreactor/protocol/http11_parser.d dsss.last test/chatclient test/chatserver test/longtest |
| diffstat | 19 files changed, 3746 insertions(+), 312 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/core/AsyncVat.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,171 @@ +/******************************************************************************* + + copyright: Copyright (c) 2008 Rick Richardson. All rights reserved + + license: BSD style: $(LICENSE) + + version: Initial release v0.1 : May 2008 + + author: Rick Richardson + +*******************************************************************************/ + +module dreactor.core.Vat; + +import tango.io.selector.Selector; +import tango.io.selector.model.ISelector; +import tango.core.Exception; +import tango.core.Thread; +import tango.core.Atomic; +import tango.util.collection.LinkSeq; +import tango.util.log.Log; + +import dreactor.transport.AsyncSocketConduit; +import dreactor.core.Dispatcher; +import dreactor.util.ThreadSafeQueue; + +Logger log; + +enum : int {CLOSE = -2, UNREGISTER = -1, REMAIN = 0, REGISTER = 1, REREGISTER = 2}; + +static char[] version_string = "Vat.d 0.1 2008-05-31"; + +class Vat +{ +private + Thread thread; + bool running; + Atomic!(int) pending; + + ThreadSafeQueue!(Dispatcher) freshList; + ThreadSafeQueue!(Dispatcher) remList; +public + this() + { + freshList = new ThreadSafeQueue!(Dispatcher); + remList = new ThreadSafeQueue!(Dispatcher); + log = Log.lookup("dreactor.core.Vat"); + } + + void run() + { + running = true; + thread = new Thread(&eventLoop); + thread.start(); + } + + void exit() + { + running = false; + } + + void wait() + { + thread.join(); + } + + bool addConnection(Dispatcher handler) + { + log.trace("adding handler"); + return freshList.push(handler); + } + + bool remConnection(Dispatcher handler) + { + return remList.push(handler); + } + +private + void eventLoop() + { + auto selector = new Selector(); + selector.open(); + do + { + auto eventCount = selector.select(0.01); + + if (eventCount > 0) + { + // process events + foreach (SelectionKey key; selector.selectedSet()) + { + if (key.isReadable()) + { + // incoming data + log.trace("Read event fired"); + auto conn = cast(Dispatcher) key.attachment; + if ( Dispatcher.State.listening == conn.getState() ) + conn.handleConnection(conn.transport, &addConnection); + else + processReturn(conn.handleIncoming(), selector, conn); + } + else if (key.isWritable()) + { + log.trace("Write event fired"); + auto conn = cast(Dispatcher) key.attachment; + processReturn(conn.handleOutgoing(), selector, conn); + } + else if (key.isHangup()) + { + log.trace("Hangup event fired"); + auto conn = cast(Dispatcher) key.attachment; + processReturn(conn.handleDisconnect(), selector, conn); + } + else if (key.isError() || key.isInvalidHandle()) + { + log.trace("Error event fired"); + // error, close connection + auto conn = cast(Dispatcher) key.attachment; + conn.handleError(&remConnection); + } + } + } + else if (eventCount == 0) + { + /* can't think of anything useful to do here. */ + } + else + { + log.error("Selector.select returned {}", eventCount); + } + //add Conduits to listener + freshList.processAll( (ref Dispatcher h) + { + selector.register(h.transport, h.events(), h); + return 1; + }); + remList.processAll( (ref Dispatcher h) + { + selector.unregister(h.transport); + return 1; + }); + + } while (running) + + } + + void processReturn(int result, Selector s, Dispatcher h) + { + switch(result) + { + case CLOSE: + s.unregister(h.transport); + h.transport.detach(); + break; + case UNREGISTER: + s.unregister(h.transport); + break; + case REMAIN: + //this space intentially left blank + break; + case REGISTER: + s.register(h.transport, h.events(), h); + break; + case REREGISTER: + s.register(h.transport, h.events(), h); + break; + default: + log.error("processReturn: unknown return value"); + } + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/core/Dispatcher.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,387 @@ +module dreactor.core.Dispatcher; + +import tango.io.selector.model.ISelector; +import tango.util.collection.CircularSeq; +import tango.net.Socket; +public import tango.core.Exception; +import dreactor.transport.AsyncSocketConduit; + +import tango.util.log.Log; +import tango.util.log.Config; + +alias bool delegate(Dispatcher) RegisterD; + +alias int delegate(Dispatcher) IncomingHandlerD; +alias int delegate(Dispatcher) OutgoingHandlerD; +alias int delegate(Dispatcher, RegisterD) ErrorHandlerD; +alias int delegate(Dispatcher) DisconnectHandlerD; +alias int delegate(Conduit, RegisterD) ConnectHandlerD; + +alias int function(Dispatcher) IncomingHandlerF; +alias int function(Dispatcher) OutgoingHandlerF; +alias int function(Dispatcher, RegisterD) ErrorHandlerF; +alias int function(Dispatcher) DisconnectHandlerF; +alias int function(Conduit, RegisterD) ConnectHandlerF; + + +/****************************************************************************** + Dispatcher object. To be used by the SelectLoop to manage callbacks + for events. It may also be used to buffer data inbetween requests. + These can be populated passed to a SelectLoop directly by the end user, + or may be managed by a chosen Protocol. +******************************************************************************/ +class Dispatcher +{ + public + enum State { init, connected, listening, idle, closing }; + + /************************************************************************** + + Standard Ctor, takes a transport_ + + **************************************************************************/ + this (Conduit trans, bool listener = false) + { + transport_ = trans; + ibuf_len = 0; + i_offset = 0; + o_offset = 0; + out_buffers = new CircularSeq!(char[]); + log = Log.lookup("dreactor.core.Dispatcher"); + } + + /********************************************************************** + + Setters for the handlers. These are set by the Protocols as well + + **********************************************************************/ + + void setIncomingHandler(IncomingHandlerD hand) + { + inD = hand; + inF = null; + } + + void setIncomingHandler(IncomingHandlerF hand) + { + inF = hand; + inD = null; + } + + void setOutgoingHandler(OutgoingHandlerD hand) + { + outD = hand; + outF = null; + } + + void setOutgoingHandler(OutgoingHandlerF hand) + { + outF = hand; + outD = null; + } + + void setErrorHandler(ErrorHandlerD hand) + { + errD = hand; + errF = null; + } + + void setErrorHandler(ErrorHandlerF hand) + { + errF = hand; + errD = null; + } + + void setDisconnectHandler(DisconnectHandlerD hand) + { + disD = hand; + disF = null; + } + + void setDisconnectHandler(DisconnectHandlerF hand) + { + disF = hand; + disD = null; + } + + void setConnectHandler(ConnectHandlerD hand) + { + conD = hand; + conF = null; + } + + void setConnectHandler(ConnectHandlerF hand) + { + conF = hand; + conD = null; + } + + /********************************************************************** + + Handlers to be called by the SelectLoop when events occur + + **********************************************************************/ + int handleIncoming() + { + if (inD !is null) + return inD(this); + else if (inF !is null) + return inF(this); + else + throw new Exception("no Incoming handler set"); + } + + int handleOutgoing() + { + if (outD !is null) + return outD(this); + else if (outF !is null) + return outF(this); + else + throw new Exception("no Outgoing handler set"); + } + + int handleError(RegisterD reg) + { + if (errD !is null) + return errD(this, reg); + else if (errF !is null) + return errF(this, reg); + } + + int handleDisconnect() + { + if (disD !is null) + return disD(this); + else if (disF !is null) + return disF(this); + } + + int handleConnection(Conduit cond, RegisterD reg ) + { + if (conD !is null) + { + return conD(cond, reg); + } + else if (conF !is null) + { + return conF(cond, reg); + } + } + + /************************************************************************** + + Sending / Receiving helpers + + **************************************************************************/ + + /************************************************************************** + + appendOutBuffer + + Adds an outgoing buffer to the list. This returns true if the list + was empty, indicating that the handler should be registered with the + SelectLoop. If it returns false, it was probably already registered. + + **************************************************************************/ + synchronized bool appendOutBuffer(char[] outbuf) + { + out_buffers.append(outbuf); + out_buffers_len++; + if (out_buffers_len == 1) + return true; + else + return false; + } + + /************************************************************************** + + addOffset + Use this function to update the offset position after a successful data + send. This not only manages the current offset, but will update the + out buffer chain if necessary. + + Returns: false if there is nothing left to send, true if there is. + + **************************************************************************/ + synchronized bool addOffset(int off) + in + { + assert(out_buffers_len > 0); + } + body + { + char[] hd = out_buffers.head(); + if ((off + o_offset) >= hd.length) + { + out_buffers.removeHead(); + o_offset = 0; + out_buffers_len--; + return (out_buffers_len > 0); + } + else + o_offset += off; + return true; + } + + /************************************************************************** + + char[] nextBuffer + + Returns a slice of the current outbound buffer, returns a char[] pointing + to null if there is no current outbound buffer + + **************************************************************************/ + synchronized char[] nextBuffer() + { + if (out_buffers_len < 1) + { + return null; + } + + return out_buffers.head()[o_offset .. $]; + } + + /************************************************************************** + + listen + Enable listening on the socket attached to this connectionhandler + + **************************************************************************/ + int listen(IPv4Address addr) + { + (cast(AsyncSocketConduit)transport_).bind(addr).listen(); + state_ = State.listening; + return 0; + } + + Conduit transport() + { + return transport_; + } + /************************************************************************** + + Configuration functions + + **************************************************************************/ + Event events() + { + return events_; + } + void events(Event e) + { + events_ = e; + } + void addEvent(Event e) + { + events_ |= e; + } + void remEvent(Event e) + { + events_ &= ~e; + } + + State getState() {return state_;} + + /* + connection handlers are left out of this because + this method is used by the listener socket to pass + on its handlers to the accepted socket. An accepted + socket will generally do different things onConnection + */ + void setHandlers(Dispatcher other) + { + inD = other.inD; + outD = other.outD; + errD = other.errD; + disD = other.disD; + inF = other.inF; + outF = other.outF; + errF = other.errF; + disF = other.disF; + } + + /************************************************************************** + + Freelist allocators and deallocators + + **************************************************************************/ + static synchronized Dispatcher New(Conduit tran, Dispatcher other = null) + { + Dispatcher hand; + if (freelist) + { + hand = freelist; + freelist = hand.next; + hand.transport_ = tran; + } + else + hand = new Dispatcher(tran); + + if (!(other is null)) + { + hand.setHandlers(other); + } + return hand; + } + + static synchronized void Delete(Dispatcher hand) + { + hand.next = freelist; + freelist = hand.initialize(); + } + +private + + char[] in_buffer; + CircularSeq!(char[]) out_buffers; + int out_buffers_len; + int ibuf_len; + int i_offset; + int o_offset; + Logger log; + + package Conduit transport_; + State state_; + Event events_; + IncomingHandlerD inD; + OutgoingHandlerD outD; + ErrorHandlerD errD; + DisconnectHandlerD disD; + ConnectHandlerD conD; + + IncomingHandlerF inF; + OutgoingHandlerF outF; + ErrorHandlerF errF; + DisconnectHandlerF disF; + ConnectHandlerF conF; + + static Dispatcher freelist; + Dispatcher next; + + /************************************************************************** + Copy ctor, creates a new Dispatcher using the settings + of an existing handler. + **************************************************************************/ + Dispatcher initialize() + { + transport_ = null; + state_ = State.init; + ibuf_len = 0; + i_offset = 0; + o_offset = 0; + out_buffers.clear(); + inD = null; + outD = null; + errD = null; + disD = null; + conD = null; + inF = null; + outF = null; + errF = null; + disF = null; + conF = null; + return this; + } +} +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/protocol/Http11.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,43 @@ +module dreactor.protocol.Http11; + +import dreactor.protocol.RawTcp; +import dreactor.protocol.http11_parser; + +class HttpListener +{ +public + this(Vat sel, IPv4Address addr) + { + listener = new RawTCPListener(sel, IPv4Address addr); + parser = new Http11Parser(); + listener.setDataHandler(&onData); + } + + private int onData(char[] buffer) + { + parser.execute(buffer); + } + + +private + RawTCPListener listener; + Http11Parser parser; +} + +class HttpClient +{ +public + this(Vat sel) + { + client = new RawTCPClient(sel); + client.setDataHandler(&onData); + } + + private int onData() + { + + } +private + RawTCPClient client; +} +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/protocol/Http11Parser.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,1293 @@ +/* #line 1 "http11_parser.rl" */ +/** + * Copyright (c) 2005 Zed A. Shaw, + * Modified for D/DReactor by Rick Richardson, 2008 + * You can redistribute it and/or modify it under the same terms as Ruby. + */ + +alias void delegate (void *data, const char *at, size_t length) element_cb; +alias void delegate (void *data, const char *field, size_t flen, const char *value, size_t vlen) field_cb; + +class Http11Parser +{ + int cs; + size_t body_start; + int content_len; + size_t nread; + size_t mark; + size_t field_start; + size_t field_len; + size_t query_start; + + void *data; + + field_cb http_field; + element_cb request_method; + element_cb request_uri; + element_cb fragment; + element_cb request_path; + element_cb query_string; + element_cb http_version; + element_cb header_done; + +/* + * capitalizes all lower-case ASCII characters, + * converts dashes to underscores. + */ +private void snake_upcase_char(char *c) +{ + if (*c >= 'a' && *c <= 'z') + *c &= ~0x20; + else if (*c == '-') + *c = '_'; +} + +//#define LEN(AT, FPC) (FPC - buffer - parser->AT) +private int LEN(char* at, char* fpc) +{ + return (fpc - buffer.ptr - at); +} + +//#define MARK(M,FPC) (parser->M = (FPC) - buffer) +private void MARK(size_t* item, char* fpc) +{ + *item = fpc - buffer.ptr; +} + +//#define PTR_TO(F) (buffer + parser->F) +private char* PTR_TO(size_t F) +{ + return (buffer.ptr + F); +} +/** Machine **/ + +/* #line 126 "http11_parser.rl" */ + + +/** Data **/ + +/* #line 70 "Http11Parser.d" */ +static const int http_parser_start = 1; +static const int http_parser_first_final = 57; +static const int http_parser_error = 0; + +static const int http_parser_en_main = 1; + +/* #line 130 "http11_parser.rl" */ + +this () +{ + cs = 0; + +/* #line 83 "Http11Parser.d" */ + { + cs = http_parser_start; + } +/* #line 135 "http11_parser.rl" */ + + body_start = 0; + content_len = 0; + mark = 0; + nread = 0; + field_len = 0; + field_start = 0; +} + + +/** exec **/ +size_t execute(const char[] buffer, size_t off) { + const char *p, *pe; + int len = buffer.length; + + assert(off <= len && "offset past end of buffer"); + + p = buffer.ptr+off; + pe = buffer.ptr+len; + + /* assert(*pe == '\0' && "pointer does not end on NUL"); */ + assert(pe - p == len - off && "pointers aren't same distance"); + + +/* #line 112 "Http11Parser.d" */ + { + if ( p == pe ) + goto _out; + switch ( cs ) + { +case 1: + switch( (*p) ) { + case 36u: goto tr0; + case 95u: goto tr0; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto tr0; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto tr0; + } else + goto tr0; + goto st0; +st0: + goto _out0; +tr0: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st2; +st2: + if ( ++p == pe ) + goto _out2; +case 2: +/* #line 143 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st38; + case 95u: goto st38; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st38; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st38; + } else + goto st38; + goto st0; +tr2: +/* #line 85 "http11_parser.rl" */ + { + if(request_method) + request_method(data, PTR_TO(mark), LEN(mark, p)); + } + goto st3; +st3: + if ( ++p == pe ) + goto _out3; +case 3: +/* #line 170 "Http11Parser.d" */ + switch( (*p) ) { + case 42u: goto tr4; + case 43u: goto tr5; + case 47u: goto tr6; + case 58u: goto tr7; + default: break; + } + if ( (*p) < 65u ) { + if ( 45u <= (*p) && (*p) <= 57u ) + goto tr5; + } else if ( (*p) > 90u ) { + if ( 97u <= (*p) && (*p) <= 122u ) + goto tr5; + } else + goto tr5; + goto st0; +tr4: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st4; +st4: + if ( ++p == pe ) + goto _out4; +case 4: +/* #line 195 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr8; + case 35u: goto tr9; + default: break; + } + goto st0; +tr8: +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st5; +tr29: +/* #line 95 "http11_parser.rl" */ + { + if(fragment) + fragment(data, PTR_TO(mark), LEN(mark, p)); + } + goto st5; +tr39: +/* #line 112 "http11_parser.rl" */ + { + if(request_path) + request_path(data, PTR_TO(mark), LEN(mark,p)); + } +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st5; +tr50: +/* #line 100 "http11_parser.rl" */ + {MARK(query_start, p); } +/* #line 102 "http11_parser.rl" */ + { + if(query_string) + query_string(data, PTR_TO(query_start), LEN(query_start, p)); + } +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st5; +tr54: +/* #line 102 "http11_parser.rl" */ + { + if(query_string) + query_string(data, PTR_TO(query_start), LEN(query_start, p)); + } +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st5; +st5: + if ( ++p == pe ) + goto _out5; +case 5: +/* #line 258 "Http11Parser.d" */ + if ( (*p) == 72u ) + goto tr10; + goto st0; +tr10: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st6; +st6: + if ( ++p == pe ) + goto _out6; +case 6: +/* #line 270 "Http11Parser.d" */ + if ( (*p) == 84u ) + goto st7; + goto st0; +st7: + if ( ++p == pe ) + goto _out7; +case 7: + if ( (*p) == 84u ) + goto st8; + goto st0; +st8: + if ( ++p == pe ) + goto _out8; +case 8: + if ( (*p) == 80u ) + goto st9; + goto st0; +st9: + if ( ++p == pe ) + goto _out9; +case 9: + if ( (*p) == 47u ) + goto st10; + goto st0; +st10: + if ( ++p == pe ) + goto _out10; +case 10: + if ( 48u <= (*p) && (*p) <= 57u ) + goto st11; + goto st0; +st11: + if ( ++p == pe ) + goto _out11; +case 11: + if ( (*p) == 46u ) + goto st12; + if ( 48u <= (*p) && (*p) <= 57u ) + goto st11; + goto st0; +st12: + if ( ++p == pe ) + goto _out12; +case 12: + if ( 48u <= (*p) && (*p) <= 57u ) + goto st13; + goto st0; +st13: + if ( ++p == pe ) + goto _out13; +case 13: + if ( (*p) == 13u ) + goto tr18; + if ( 48u <= (*p) && (*p) <= 57u ) + goto st13; + goto st0; +tr18: +/* #line 107 "http11_parser.rl" */ + { + if(http_version) + http_version(data, PTR_TO(mark), LEN(mark, p)); + } + goto st14; +tr26: +/* #line 77 "http11_parser.rl" */ + { + if(http_field) + http_field(data, + PTR_TO(field_start), + field_len, + PTR_TO(mark), + LEN(mark, p)); + } + goto st14; +st14: + if ( ++p == pe ) + goto _out14; +case 14: +/* #line 349 "Http11Parser.d" */ + if ( (*p) == 10u ) + goto st15; + goto st0; +st15: + if ( ++p == pe ) + goto _out15; +case 15: + switch( (*p) ) { + case 13u: goto st16; + case 33u: goto tr21; + case 124u: goto tr21; + case 126u: goto tr21; + default: break; + } + if ( (*p) < 45u ) { + if ( (*p) > 39u ) { + if ( 42u <= (*p) && (*p) <= 43u ) + goto tr21; + } else if ( (*p) >= 35u ) + goto tr21; + } else if ( (*p) > 46u ) { + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto tr21; + } else if ( (*p) > 90u ) { + if ( 94u <= (*p) && (*p) <= 122u ) + goto tr21; + } else + goto tr21; + } else + goto tr21; + goto st0; +st16: + if ( ++p == pe ) + goto _out16; +case 16: + if ( (*p) == 10u ) + goto tr22; + goto st0; +tr22: +/* #line 117 "http11_parser.rl" */ + { + body_start = p - buffer + 1; + if(header_done) + header_done(data, p + 1, pe - p - 1); + if (true) goto _out57; + } + goto st57; +st57: + if ( ++p == pe ) + goto _out57; +case 57: +/* #line 402 "Http11Parser.d" */ + goto st0; +tr21: +/* #line 70 "http11_parser.rl" */ + { MARK(field_start, p); } +/* #line 71 "http11_parser.rl" */ + { snake_upcase_char((char *)p); } + goto st17; +tr23: +/* #line 71 "http11_parser.rl" */ + { snake_upcase_char((char *)p); } + goto st17; +st17: + if ( ++p == pe ) + goto _out17; +case 17: +/* #line 418 "Http11Parser.d" */ + switch( (*p) ) { + case 33u: goto tr23; + case 58u: goto tr24; + case 124u: goto tr23; + case 126u: goto tr23; + default: break; + } + if ( (*p) < 45u ) { + if ( (*p) > 39u ) { + if ( 42u <= (*p) && (*p) <= 43u ) + goto tr23; + } else if ( (*p) >= 35u ) + goto tr23; + } else if ( (*p) > 46u ) { + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto tr23; + } else if ( (*p) > 90u ) { + if ( 94u <= (*p) && (*p) <= 122u ) + goto tr23; + } else + goto tr23; + } else + goto tr23; + goto st0; +tr24: +/* #line 72 "http11_parser.rl" */ + { + field_len = LEN(field_start, p); + } + goto st18; +tr27: +/* #line 76 "http11_parser.rl" */ + { MARK(mark, p); } + goto st18; +st18: + if ( ++p == pe ) + goto _out18; +case 18: +/* #line 458 "Http11Parser.d" */ + switch( (*p) ) { + case 13u: goto tr26; + case 32u: goto tr27; + default: break; + } + goto tr25; +tr25: +/* #line 76 "http11_parser.rl" */ + { MARK(mark, p); } + goto st19; +st19: + if ( ++p == pe ) + goto _out19; +case 19: +/* #line 473 "Http11Parser.d" */ + if ( (*p) == 13u ) + goto tr26; + goto st19; +tr9: +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st20; +tr41: +/* #line 112 "http11_parser.rl" */ + { + if(request_path) + request_path(data, PTR_TO(mark), LEN(mark,p)); + } +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st20; +tr52: +/* #line 100 "http11_parser.rl" */ + {MARK(query_start, p); } +/* #line 102 "http11_parser.rl" */ + { + if(query_string) + query_string(data, PTR_TO(query_start), LEN(query_start, p)); + } +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st20; +tr56: +/* #line 102 "http11_parser.rl" */ + { + if(query_string) + query_string(data, PTR_TO(query_start), LEN(query_start, p)); + } +/* #line 90 "http11_parser.rl" */ + { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, p)); + } + goto st20; +st20: + if ( ++p == pe ) + goto _out20; +case 20: +/* #line 526 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr29; + case 37u: goto tr31; + case 60u: goto st0; + case 62u: goto st0; + case 127u: goto st0; + default: break; + } + if ( (*p) > 31u ) { + if ( 34u <= (*p) && (*p) <= 35u ) + goto st0; + } else + goto st0; + goto tr30; +tr30: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st21; +st21: + if ( ++p == pe ) + goto _out21; +case 21: +/* #line 549 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr29; + case 37u: goto st22; + case 60u: goto st0; + case 62u: goto st0; + case 127u: goto st0; + default: break; + } + if ( (*p) > 31u ) { + if ( 34u <= (*p) && (*p) <= 35u ) + goto st0; + } else + goto st0; + goto st21; +tr31: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st22; +st22: + if ( ++p == pe ) + goto _out22; +case 22: +/* #line 572 "Http11Parser.d" */ + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st23; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st23; + } else + goto st23; + goto st0; +st23: + if ( ++p == pe ) + goto _out23; +case 23: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st21; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st21; + } else + goto st21; + goto st0; +tr5: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st24; +st24: + if ( ++p == pe ) + goto _out24; +case 24: +/* #line 603 "Http11Parser.d" */ + switch( (*p) ) { + case 43u: goto st24; + case 58u: goto st25; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st24; + } else if ( (*p) > 57u ) { + if ( (*p) > 90u ) { + if ( 97u <= (*p) && (*p) <= 122u ) + goto st24; + } else if ( (*p) >= 65u ) + goto st24; + } else + goto st24; + goto st0; +tr7: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st25; +st25: + if ( ++p == pe ) + goto _out25; +case 25: +/* #line 629 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr8; + case 34u: goto st0; + case 35u: goto tr9; + case 37u: goto st26; + case 60u: goto st0; + case 62u: goto st0; + case 127u: goto st0; + default: break; + } + if ( (*p) <= 31u ) + goto st0; + goto st25; +st26: + if ( ++p == pe ) + goto _out26; +case 26: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st27; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st27; + } else + goto st27; + goto st0; +st27: + if ( ++p == pe ) + goto _out27; +case 27: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st25; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st25; + } else + goto st25; + goto st0; +tr6: +/* #line 67 "http11_parser.rl" */ + {MARK(mark, p); } + goto st28; +st28: + if ( ++p == pe ) + goto _out28; +case 28: +/* #line 677 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr39; + case 34u: goto st0; + case 35u: goto tr41; + case 37u: goto st29; + case 59u: goto tr43; + case 60u: goto st0; + case 62u: goto st0; + case 63u: goto tr44; + case 127u: goto st0; + default: break; + } + if ( (*p) <= 31u ) + goto st0; + goto st28; +st29: + if ( ++p == pe ) + goto _out29; +case 29: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st30; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st30; + } else + goto st30; + goto st0; +st30: + if ( ++p == pe ) + goto _out30; +case 30: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st28; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st28; + } else + goto st28; + goto st0; +tr43: +/* #line 112 "http11_parser.rl" */ + { + if(request_path) + request_path(data, PTR_TO(mark), LEN(mark,p)); + } + goto st31; +st31: + if ( ++p == pe ) + goto _out31; +case 31: +/* #line 730 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr8; + case 34u: goto st0; + case 35u: goto tr9; + case 37u: goto st32; + case 60u: goto st0; + case 62u: goto st0; + case 63u: goto st34; + case 127u: goto st0; + default: break; + } + if ( (*p) <= 31u ) + goto st0; + goto st31; +st32: + if ( ++p == pe ) + goto _out32; +case 32: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st33; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st33; + } else + goto st33; + goto st0; +st33: + if ( ++p == pe ) + goto _out33; +case 33: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st31; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st31; + } else + goto st31; + goto st0; +tr44: +/* #line 112 "http11_parser.rl" */ + { + if(request_path) + request_path(data, PTR_TO(mark), LEN(mark,p)); + } + goto st34; +st34: + if ( ++p == pe ) + goto _out34; +case 34: +/* #line 782 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr50; + case 34u: goto st0; + case 35u: goto tr52; + case 37u: goto tr53; + case 60u: goto st0; + case 62u: goto st0; + case 127u: goto st0; + default: break; + } + if ( (*p) <= 31u ) + goto st0; + goto tr51; +tr51: +/* #line 100 "http11_parser.rl" */ + {MARK(query_start, p); } + goto st35; +st35: + if ( ++p == pe ) + goto _out35; +case 35: +/* #line 804 "Http11Parser.d" */ + switch( (*p) ) { + case 32u: goto tr54; + case 34u: goto st0; + case 35u: goto tr56; + case 37u: goto st36; + case 60u: goto st0; + case 62u: goto st0; + case 127u: goto st0; + default: break; + } + if ( (*p) <= 31u ) + goto st0; + goto st35; +tr53: +/* #line 100 "http11_parser.rl" */ + {MARK(query_start, p); } + goto st36; +st36: + if ( ++p == pe ) + goto _out36; +case 36: +/* #line 826 "Http11Parser.d" */ + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st37; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st37; + } else + goto st37; + goto st0; +st37: + if ( ++p == pe ) + goto _out37; +case 37: + if ( (*p) < 65u ) { + if ( 48u <= (*p) && (*p) <= 57u ) + goto st35; + } else if ( (*p) > 70u ) { + if ( 97u <= (*p) && (*p) <= 102u ) + goto st35; + } else + goto st35; + goto st0; +st38: + if ( ++p == pe ) + goto _out38; +case 38: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st39; + case 95u: goto st39; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st39; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st39; + } else + goto st39; + goto st0; +st39: + if ( ++p == pe ) + goto _out39; +case 39: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st40; + case 95u: goto st40; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st40; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st40; + } else + goto st40; + goto st0; +st40: + if ( ++p == pe ) + goto _out40; +case 40: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st41; + case 95u: goto st41; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st41; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st41; + } else + goto st41; + goto st0; +st41: + if ( ++p == pe ) + goto _out41; +case 41: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st42; + case 95u: goto st42; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st42; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st42; + } else + goto st42; + goto st0; +st42: + if ( ++p == pe ) + goto _out42; +case 42: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st43; + case 95u: goto st43; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st43; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st43; + } else + goto st43; + goto st0; +st43: + if ( ++p == pe ) + goto _out43; +case 43: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st44; + case 95u: goto st44; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st44; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st44; + } else + goto st44; + goto st0; +st44: + if ( ++p == pe ) + goto _out44; +case 44: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st45; + case 95u: goto st45; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st45; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st45; + } else + goto st45; + goto st0; +st45: + if ( ++p == pe ) + goto _out45; +case 45: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st46; + case 95u: goto st46; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st46; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st46; + } else + goto st46; + goto st0; +st46: + if ( ++p == pe ) + goto _out46; +case 46: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st47; + case 95u: goto st47; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st47; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st47; + } else + goto st47; + goto st0; +st47: + if ( ++p == pe ) + goto _out47; +case 47: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st48; + case 95u: goto st48; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st48; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st48; + } else + goto st48; + goto st0; +st48: + if ( ++p == pe ) + goto _out48; +case 48: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st49; + case 95u: goto st49; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st49; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st49; + } else + goto st49; + goto st0; +st49: + if ( ++p == pe ) + goto _out49; +case 49: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st50; + case 95u: goto st50; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st50; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st50; + } else + goto st50; + goto st0; +st50: + if ( ++p == pe ) + goto _out50; +case 50: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st51; + case 95u: goto st51; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st51; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st51; + } else + goto st51; + goto st0; +st51: + if ( ++p == pe ) + goto _out51; +case 51: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st52; + case 95u: goto st52; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st52; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st52; + } else + goto st52; + goto st0; +st52: + if ( ++p == pe ) + goto _out52; +case 52: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st53; + case 95u: goto st53; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st53; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st53; + } else + goto st53; + goto st0; +st53: + if ( ++p == pe ) + goto _out53; +case 53: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st54; + case 95u: goto st54; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st54; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st54; + } else + goto st54; + goto st0; +st54: + if ( ++p == pe ) + goto _out54; +case 54: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st55; + case 95u: goto st55; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st55; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st55; + } else + goto st55; + goto st0; +st55: + if ( ++p == pe ) + goto _out55; +case 55: + switch( (*p) ) { + case 32u: goto tr2; + case 36u: goto st56; + case 95u: goto st56; + default: break; + } + if ( (*p) < 48u ) { + if ( 45u <= (*p) && (*p) <= 46u ) + goto st56; + } else if ( (*p) > 57u ) { + if ( 65u <= (*p) && (*p) <= 90u ) + goto st56; + } else + goto st56; + goto st0; +st56: + if ( ++p == pe ) + goto _out56; +case 56: + if ( (*p) == 32u ) + goto tr2; + goto st0; + default: break; + } + _out0: cs = 0; goto _out; + _out2: cs = 2; goto _out; + _out3: cs = 3; goto _out; + _out4: cs = 4; goto _out; + _out5: cs = 5; goto _out; + _out6: cs = 6; goto _out; + _out7: cs = 7; goto _out; + _out8: cs = 8; goto _out; + _out9: cs = 9; goto _out; + _out10: cs = 10; goto _out; + _out11: cs = 11; goto _out; + _out12: cs = 12; goto _out; + _out13: cs = 13; goto _out; + _out14: cs = 14; goto _out; + _out15: cs = 15; goto _out; + _out16: cs = 16; goto _out; + _out57: cs = 57; goto _out; + _out17: cs = 17; goto _out; + _out18: cs = 18; goto _out; + _out19: cs = 19; goto _out; + _out20: cs = 20; goto _out; + _out21: cs = 21; goto _out; + _out22: cs = 22; goto _out; + _out23: cs = 23; goto _out; + _out24: cs = 24; goto _out; + _out25: cs = 25; goto _out; + _out26: cs = 26; goto _out; + _out27: cs = 27; goto _out; + _out28: cs = 28; goto _out; + _out29: cs = 29; goto _out; + _out30: cs = 30; goto _out; + _out31: cs = 31; goto _out; + _out32: cs = 32; goto _out; + _out33: cs = 33; goto _out; + _out34: cs = 34; goto _out; + _out35: cs = 35; goto _out; + _out36: cs = 36; goto _out; + _out37: cs = 37; goto _out; + _out38: cs = 38; goto _out; + _out39: cs = 39; goto _out; + _out40: cs = 40; goto _out; + _out41: cs = 41; goto _out; + _out42: cs = 42; goto _out; + _out43: cs = 43; goto _out; + _out44: cs = 44; goto _out; + _out45: cs = 45; goto _out; + _out46: cs = 46; goto _out; + _out47: cs = 47; goto _out; + _out48: cs = 48; goto _out; + _out49: cs = 49; goto _out; + _out50: cs = 50; goto _out; + _out51: cs = 51; goto _out; + _out52: cs = 52; goto _out; + _out53: cs = 53; goto _out; + _out54: cs = 54; goto _out; + _out55: cs = 55; goto _out; + _out56: cs = 56; goto _out; + + _out: {} + } +/* #line 159 "http11_parser.rl" */ + + //parser->cs = cs; + nread += p - (buffer.ptr + off); + + assert(p <= pe && "buffer overflow after parsing execute"); + assert(nread <= len && "nread longer than length"); + assert(body_start <= len && "body starts after buffer end"); + assert(mark < len && "mark is after buffer end"); + assert(field_len <= len && "field has length longer than whole buffer"); + assert(field_start < len && "field starts after buffer end"); + + return(nread); +} + +int finish() +{ + if (has_error() ) { + return -1; + } else if (is_finished() ) { + return 1; + } else { + return 0; + } +} + +bool has_error() { + return cs == http_parser_error; +} + +bool is_finished() { + return cs >= http_parser_first_final; +} + +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/protocol/RawTcp.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,345 @@ +module dreactor.protocol.RawTcp; + +import tango.io.Conduit; +import tango.io.selector.model.ISelector; +import tango.net.Socket; +import tango.util.collection.CircularSeq; +import tango.util.log.Log; +import tango.util.log.Config; + +import dreactor.transport.AsyncSocketConduit; +import dreactor.core.Vat; +import dreactor.core.Dispatcher; + +/****************************************************************************** + + Basic TCP server or client routines for sending raw data. + +******************************************************************************/ +class RawTCPListener +{ +public + this(Dispatcher mgr, Vat sel, IPv4Address addr) + { + manager = mgr; + mgr.events(Event.Read); + mgr.setOutgoingHandler(&RawTCPHandler.onSend); + mgr.setIncomingHandler(&onReceive); + mgr.setConnectHandler(&accept); + mgr.setErrorHandler(&RawTCPHandler.onError); + mgr.setDisconnectHandler(&RawTCPHandler.onHangup); + mgr.listen(addr); + + sel.addConnection(mgr); + vat = sel; + log = Log.lookup("dreactor.protocol.RawTcpServer"); + log.info("log initialized"); + children = new CircularSeq!(Dispatcher); + } + this(Vat sel, IPv4Address addr) + { + AsyncSocketConduit cond = new AsyncSocketConduit; + cond.socket().setAddressReuse(true); + Dispatcher lh = new Dispatcher(cond, true); + this(lh, sel, addr); + } + + ~this() + { + close(); + } + + int accept(Conduit cond, RegisterD reg) + { + AsyncSocketConduit newcond = new AsyncSocketConduit; + (cast(AsyncSocketConduit)cond).socket().accept(newcond.socket); + Dispatcher h = Dispatcher.New(newcond, manager); + h.events(Event.Read); + vat.addConnection(h); + children.append(h); + log.info("accepted new connection"); + return 0; + } + + int broadcast(char[] outbuf, Dispatcher[] excluded = null) + { + foreach(Dispatcher h; children) + { + if (excluded && excluded.includes(h)) + continue; + if (h.appendOutBuffer(outbuf)) + { + h.addEvent(Event.Write); + vat.addConnection(h); + } + } + return 0; + } + + /************************************************************************** + + send + User-called function to send data to the counterpart at the other + end of the connection. This sets up the connection manager to send + data as the socket becomes free. + + **************************************************************************/ + int send(Dispatcher d, char[] outbuf, IPv4Address addr = null) + { + if (d.appendOutBuffer(outbuf)) + { + d.addEvent(Event.Write); + if (!vat.addConnection(d)) + { + log.error("unable to register mgr"); + } + } + return 0; + } + + /************************************************************************** + + receive + IncomingHandlerD + Default incoming data handler. Should be replaced with something useful. + + **************************************************************************/ + int onReceive(Dispatcher h) + { + Logger log = Log.lookup("Handlers.onReceive"); + + char inbuf[8192]; + int amt; + if((amt = h.transport.read(inbuf)) > 0) + { + if (dataHandler) + dataHandler(inbuf[0 .. amt], h); + else + log.info("Received {} byte Buffer: {}", amt, inbuf[0 .. amt]); + } + else + { + if (amt == 0) + { + children.remove(h); + (cast(AsyncSocketConduit) h.transport).shutdown(); + return CLOSE; + } + log.error("Received no data, err = {}", amt); + } + return REMAIN; + } + + void close() + { + foreach(Dispatcher d; children) + { + (cast(AsyncSocketConduit)d.transport).shutdown(); + (cast(AsyncSocketConduit)d.transport).detach(); + } + (cast(AsyncSocketConduit)manager.transport).shutdown(); + (cast(AsyncSocketConduit)manager.transport).detach(); + + } + + void setDataHandler(void delegate(char[], Dispatcher) h) + { + dataHandler = h; + } + +private + Vat vat; + CircularSeq!(Dispatcher) children; + Dispatcher manager; + Logger log; + RawTCPHandler h; + void delegate(char[], Dispatcher) dataHandler; +} + +class RawTCPClient +{ + +public + this(Dispatcher mgr, Vat sel, Event evts = Event.Read) + { + manager = mgr; + manager.events(evts); + connected = false; + mgr.setOutgoingHandler(&RawTCPHandler.onSend); + mgr.setIncomingHandler(&onReceive); + mgr.setErrorHandler(&RawTCPHandler.onError); + mgr.setDisconnectHandler(&RawTCPHandler.onHangup); + vat = sel; + log = Log.lookup("dreactor.protocol.RawTcpClient"); + } + + this(Vat sel, Event evts = Event.Read) + { + AsyncSocketConduit clcond = new AsyncSocketConduit; + Dispatcher ch = new Dispatcher(clcond); + this(ch, sel, evts); + } + + ~this() + { + (cast(AsyncSocketConduit)manager.transport).shutdown(); + (cast(AsyncSocketConduit)manager.transport).detach(); + } + + int connect(IPv4Address addr) + { + (cast(AsyncSocketConduit) manager.transport()).connect(addr); + vat.addConnection(manager); + connected = true; + log.info("connected to {}", addr); + return 0; + } + + /************************************************************************** + + send + User-called function to send data to the counterpart at the other + end of the connection. This sets up the connection manager to send + data as the socket becomes free. + + **************************************************************************/ + int send(char[] outbuf, IPv4Address addr = null) + { + if (!connected) + { + log.info("send: not connected, connecting"); + if (addr !is null) + { + if (0 > connect(addr)) + { + log.error("send: unable to connect"); + return -1; + } + } + } + if (manager.appendOutBuffer(outbuf)) + { + manager.addEvent(Event.Write); + if (!vat.addConnection(manager)) + { + log.error("unable to register mgr"); + } + } + return 0; + } + + /************************************************************************** + + receive + IncomingHandlerD + Default incoming data handler. Should be replaced with something useful. + + **************************************************************************/ + public int onReceive(Dispatcher h) + { + Logger log = Log.lookup("Handlers.onReceive"); + + char inbuf[8192]; + int amt; + if((amt = h.transport.read(inbuf)) > 0) + { + if (dataHandler) + dataHandler(inbuf[0 .. amt], h); + else + log.info("Received {} byte Buffer: {}", amt, inbuf[0 .. amt]); + } + else + { + if (amt == 0) + { + return CLOSE; + } + log.error("Received no data, err = {}", amt); + } + return REMAIN; + } + + void setDataHandler(void delegate(char[], Dispatcher) h) + { + dataHandler = h; + } +private + void delegate(char[], Dispatcher) dataHandler; + Dispatcher manager; + Vat vat; + bool connected; + Logger log; + RawTCPHandler h; +} + + +/****************************************************************************** + + Default Event handlers common to both listener/clients + +******************************************************************************/ +struct RawTCPHandler +{ + /************************************************************************** + + onSend + OutgoingHandlerD + To be registered as the response to socket writable event. + Sends data, returns amount sent. Unregisters Handler for sending + if there is no more data left to send. + + ***************************************************************************/ + public static int onSend(Dispatcher h) + { + Logger log = Log.lookup("Handlers.onSend"); + + char[] outbuf = h.nextBuffer(); + if (outbuf !is null) + { + int sent = h.transport.write(outbuf); + if (sent > 0) + { + if (! h.addOffset(sent)) + { + h.remEvent(Event.Write); + return REREGISTER; + } + } + else if (sent == AsyncSocketConduit.Eof) + { + log.error("Select said socket was writable, but sent 0 bytes"); + } + else + { + log.error("Socket send return ERR"); + } + return REMAIN; + } + else + { + h.remEvent(Event.Write); + return REREGISTER; + } + } + + static int onHangup(Dispatcher d) + { + return UNREGISTER; + } + + static int onError(Dispatcher d, RegisterD unreg) + { + return CLOSE; + } + +} + +bool includes(Dispatcher[] haystack, Dispatcher needle) +{ + foreach(Dispatcher h; haystack) + { + if (h is needle) + return true; + } + return false; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/protocol/RawUdp.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,133 @@ +module dreactor.protocol.Raw; + +import tango.io.Conduit; +import tango.io.selector.model.ISelector; +import dreactor.core.AsyncConduit; +import dreactor.core.SelectLoop; +import dreactor.core.ConnectionHandler; +import tango.util.collection.CircularSeq; +import tango.util.log.Log; +import tango.util.log.Configurator; + +Logger log = Log.getLogger("dreactor.core.SelectLoop"); + +/****************************************************************************** + + Basic TCP server or client routines for sending raw data. + +******************************************************************************/ +class RawListener +{ +public + + this(ConnectionHandler mgr, SelectLoop sel) + { + manager = mgr; + mgr.events(Event.Read); + sel.addConnection(mgr); + select = sel; + children = CircularSeq!(ConnectionHandler); + Configurator(); + } + + int accept(Conduit cond) + { + AsyncConduit newcond = new AsyncConduit; + cond.socket().accept(newcond.socket); + ConnectionHandler h = ConnectionHandler.New(manager); + mgr.events(Event.Read); + select.addConnection(mgr); + children.append(mgr); + } + + bool broadcast(char[] outbuf) + { + foreach(ConnectionHandler h; children) + { + if (h.appendBuffer(outbuf)) + { + h.addEvent(Event.Write); + select.addConnection(h); + } + } + } + + void close() + { + + } + + /************************************************************************** + + send + OutgoingHandlerD + To be registered as the response to socket writable event. + Sends data, returns amount sent. Unregisters Handler for sending + if there is no more data left to send. + + ***************************************************************************/ + int send(ConnectionHandler h, RegisterD reg) + { + char[] outbuf = h.nextBuffer(); + if (!outbuf is null) + { + int sent = h.transport.write(outbuf); + if (sent > 0) + { + if (! h.addOffset(sent)) + { + h.removeEvent(Event.write); + reg(h); + } + } + else if (sent == EOF) + { + // EAGAIN ? probably shouldn't have happened. + } + else + { + log.error("Socket send return ERR"); + } + return sent; + } + return 0; + } + + /************************************************************************** + + receive + IncomingHandlerD + Default incoming data handler. Should be replaced with something useful. + + **************************************************************************/ + int receive(ConnectionHandler h, RegisterD reg) + { + char inbuf[8192]; + auto format = Log.format; + if(h.transport.read(inbuf) > 0) + log.info(format("Received Buffer: {}", inbuf)); + } + +private + ConnectionHandler manager; + SelectLoop select; + CircularSeq!(ConnectionHandler) children; +} + +class RawClient +{ +public + this(ConnectionHandler mgr, SelectLoop sel) + { + manager = mgr; + mgr.events(Event.Read); + sel.addConnection(mgr); + select = sel; + } + + + +private + ConnectionHandler manager; + SelectLoop select; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/protocol/http11_parser.rl Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,192 @@ +/** + * Copyright (c) 2005 Zed A. Shaw, + * Modified for D/DReactor by Rick Richardson, 2008 + * You can redistribute it and/or modify it under the same terms as Ruby. + */ + +alias void delegate (void *data, const char *at, size_t length) element_cb; +alias void delegate (void *data, const char *field, size_t flen, const char *value, size_t vlen) field_cb; + +class Http11Parser +{ + int cs; + size_t body_start; + int content_len; + size_t nread; + size_t mark; + size_t field_start; + size_t field_len; + size_t query_start; + + void *data; + + field_cb http_field; + element_cb request_method; + element_cb request_uri; + element_cb fragment; + element_cb request_path; + element_cb query_string; + element_cb http_version; + element_cb header_done; + +/* + * capitalizes all lower-case ASCII characters, + * converts dashes to underscores. + */ +private void snake_upcase_char(char *c) +{ + if (*c >= 'a' && *c <= 'z') + *c &= ~0x20; + else if (*c == '-') + *c = '_'; +} + +//#define LEN(AT, FPC) (FPC - buffer - parser->AT) +private int LEN(char* at, char* fpc) +{ + return (fpc - buffer.ptr - at); +} + +//#define MARK(M,FPC) (parser->M = (FPC) - buffer) +private void MARK(size_t* item, char* fpc) +{ + *item = fpc - buffer.ptr; +} + +//#define PTR_TO(F) (buffer + parser->F) +private char* PTR_TO(size_t F) +{ + return (buffer.ptr + F); +} +/** Machine **/ + +%%{ + + machine http_parser; + + action mark {MARK(mark, fpc); } + + + action start_field { MARK(field_start, fpc); } + action snake_upcase_field { snake_upcase_char((char *)fpc); } + action write_field { + field_len = LEN(field_start, fpc); + } + + action start_value { MARK(mark, fpc); } + action write_value { + if(http_field) + http_field(data, + PTR_TO(field_start), + field_len, + PTR_TO(mark), + LEN(mark, fpc)); + } + action request_method { + if(request_method) + request_method(data, PTR_TO(mark), LEN(mark, fpc)); + } + + action request_uri { + if(request_uri) + request_uri(data, PTR_TO(mark), LEN(mark, fpc)); + } + + action fragment { + if(fragment) + fragment(data, PTR_TO(mark), LEN(mark, fpc)); + } + + action start_query {MARK(query_start, fpc); } + + action query_string { + if(query_string) + query_string(data, PTR_TO(query_start), LEN(query_start, fpc)); + } + + action http_version { + if(http_version) + http_version(data, PTR_TO(mark), LEN(mark, fpc)); + } + + action request_path { + if(request_path) + request_path(data, PTR_TO(mark), LEN(mark,fpc)); + } + + action done { + body_start = fpc - buffer + 1; + if(header_done) + header_done(data, fpc + 1, pe - fpc - 1); + fbreak; + } + + include http_parser_common "http11_parser_common.rl"; + +}%% + +/** Data **/ +%% write data; + +this () +{ + cs = 0; + %% write init; + + body_start = 0; + content_len = 0; + mark = 0; + nread = 0; + field_len = 0; + field_start = 0; +} + + +/** exec **/ +size_t execute(const char[] buffer, size_t off) { + const char *p, *pe; + int len = buffer.length; + + assert(off <= len && "offset past end of buffer"); + + p = buffer.ptr+off; + pe = buffer.ptr+len; + + /* assert(*pe == '\0' && "pointer does not end on NUL"); */ + assert(pe - p == len - off && "pointers aren't same distance"); + + %% write exec; + + //parser->cs = cs; + nread += p - (buffer.ptr + off); + + assert(p <= pe && "buffer overflow after parsing execute"); + assert(nread <= len && "nread longer than length"); + assert(body_start <= len && "body starts after buffer end"); + assert(mark < len && "mark is after buffer end"); + assert(field_len <= len && "field has length longer than whole buffer"); + assert(field_start < len && "field starts after buffer end"); + + return(nread); +} + +int finish() +{ + if (has_error() ) { + return -1; + } else if (is_finished() ) { + return 1; + } else { + return 0; + } +} + +bool has_error() { + return cs == http_parser_error; +} + +bool is_finished() { + return cs >= http_parser_first_final; +} + +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/transport/AsyncSocketConduit.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,227 @@ +/******************************************************************************* + +copyright: Copyright (c) 2004 Kris Bell. All rights reserved + +license: BSD style: $(LICENSE) + +version: Mar 2004 : Initial release +version: Jan 2005 : RedShodan patch for timeout query +version: Dec 2006 : Outback release + +author: Kris, modified by Rick Richardson (May 2008) + + *******************************************************************************/ + +module dreactor.transport.AsyncSocketConduit; + +private import tango.time.Time; + +public import tango.io.Conduit; + +private import tango.net.Socket; + +/******************************************************************************* + + A wrapper around the bare Socket to implement the IConduit abstraction + and add socket-specific functionality specifically for multiplexing via + poll and the ilk. + + AsyncSocketConduit data-transfer is typically performed in conjunction with + an IBuffer, but can happily be handled directly using void array where + preferred + + *******************************************************************************/ + +class AsyncSocketConduit : Conduit +{ + package Socket socket_; + + /*********************************************************************** + + Create a streaming Internet Socket + + ***********************************************************************/ + + this () + { + this (SocketType.STREAM, ProtocolType.TCP); + } + + /*********************************************************************** + + Create an Internet Socket. Used by subclasses and by + ServerSocket; the latter via method allocate() below + + ***********************************************************************/ + + protected this (SocketType type, ProtocolType protocol, bool create=true) + { + socket_ = new Socket (AddressFamily.INET, type, protocol, create); + socket_.blocking(false); + } + + /*********************************************************************** + + Return the name of this device + + ***********************************************************************/ + + override char[] toString() + { + return socket.toString; + } + + /*********************************************************************** + + Return the socket wrapper + + ***********************************************************************/ + + Socket socket () + { + return socket_; + } + + /*********************************************************************** + + Return a preferred size for buffering conduit I/O + + ***********************************************************************/ + + override uint bufferSize () + { + return 1024 * 8; + } + + /*********************************************************************** + + Models a handle-oriented device. + + TODO: figure out how to avoid exposing this in the general + case + + ***********************************************************************/ + + override Handle fileHandle () + { + return cast(Handle) socket_.fileHandle; + } + + /*********************************************************************** + + Is this socket still alive? + + ***********************************************************************/ + + override bool isAlive () + { + return socket_.isAlive; + } + + /*********************************************************************** + + Connect to the provided endpoint + + ***********************************************************************/ + + AsyncSocketConduit connect (Address addr) + { + socket_.connect (addr); + return this; + } + + /*********************************************************************** + + Bind the socket. This is typically used to configure a + listening socket (such as a server or multicast socket). + The address given should describe a local adapter, or + specify the port alone (ADDR_ANY) to have the OS assign + a local adapter address. + + ***********************************************************************/ + + AsyncSocketConduit bind (Address address) + { + socket_.bind (address); + return this; + } + + /************************************************************************** + + Enable the socket for listening + + **************************************************************************/ + AsyncSocketConduit listen(int backlog = 255) + { + socket_.listen(backlog); + return this; + } + + /*********************************************************************** + + Inform other end of a connected socket that we're no longer + available. In general, this should be invoked before close() + is invoked + + The shutdown function shuts down the connection of the socket: + + - stops receiving data for this socket. If further data + arrives, it is rejected. + + - stops trying to transmit data from this socket. Also + discards any data waiting to be sent. Stop looking for + acknowledgement of data already sent; don't retransmit + if any data is lost. + + ***********************************************************************/ + + AsyncSocketConduit shutdown () + { + socket_.shutdown (SocketShutdown.BOTH); + return this; + } + + /*********************************************************************** + + Release this AsyncSocketConduit + + Note that one should always disconnect a AsyncSocketConduit + under normal conditions, and generally invoke shutdown + on all connected sockets beforehand + + ***********************************************************************/ + + override void detach () + { + socket_.detach; + } + + /*********************************************************************** + + Read content from the socket. + + Returns the number of bytes read from the socket, or + IConduit.Eof where there's no more content available + + Return IConduit.Eof if there is an error with the socket. + + ***********************************************************************/ + override uint read (void[] dst) + { + // invoke the actual read op + return socket_.receive(dst); + } + + + /*********************************************************************** + + Callback routine to write the provided content to the + socket. + ***********************************************************************/ + + override uint write (void[] src) + { + return socket_.send (src); + } +} +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/asyncdreactor/util/ThreadSafeQueue.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,99 @@ +module dreactor.util.ThreadSafeQueue; + +import tango.util.collection.CircularSeq; +import tango.core.Atomic; + +import tango.util.log.Log; +import tango.util.log.Config; +/****************************************************************************** + + ThreadSafeQueue + Queue that is probably thread safe. It acts as a job queue, in that + you can push or pop off of the queue. Or you can processAll, which will + apply a delegate to each item, then clear the list. + +******************************************************************************/ +class ThreadSafeQueue(TYPE) +{ +public + this(int maxsz = 1000) + { + list_ = new CircularSeq!(TYPE); + maxsize_ = maxsz; + size_ = 0; + log = Log.lookup("dreactor.util.ThreadSafeQueue"); + } + + synchronized bool pop(ref TYPE t) + { + if (size_ > 0) + { + t = list_.head(); + list_.removeHead(); + size_--; + return true; + } + else + return false; + } + + synchronized bool push(TYPE t) + { + if (size_ < maxsize_) + { + list_.append(t); + size_++; + return true; + } + else + return false; + } + + synchronized int size() + { + return size_; + } + + synchronized int processAll(int delegate(ref TYPE value) dg) + { + if (0 >= size_) + return 0; + + int count = 0; + foreach(TYPE t; list_) + { + if (dg(t) < 0) + break; + ++count; + } + if (count == size_) + { + clear_(); + size_ = 0; + } + else + { + list_.removeRange(0, count); + size_ -= count; + } + return count; + } + + synchronized void clear() + { + clear_(); + } + +private + + void clear_() + { + list_.clear(); + size_ = 0 ; + } + + int maxsize_; + int size_; + Logger log; + CircularSeq!(TYPE) list_; +}
--- a/dreactor/core/Dispatcher.d Fri Aug 01 16:30:45 2008 -0400 +++ b/dreactor/core/Dispatcher.d Tue Aug 12 16:59:56 2008 -0400 @@ -1,4 +1,4 @@ -module dreactor.core.Dispatcher; +module dreactor.protocol.Dispatcher; import tango.io.selector.model.ISelector; import tango.util.collection.CircularSeq; @@ -9,173 +9,58 @@ import tango.util.log.Log; import tango.util.log.Config; -alias bool delegate(Dispatcher) RegisterD; - -alias int delegate(Dispatcher) IncomingHandlerD; -alias int delegate(Dispatcher) OutgoingHandlerD; -alias int delegate(Dispatcher, RegisterD) ErrorHandlerD; -alias int delegate(Dispatcher) DisconnectHandlerD; -alias int delegate(Conduit, RegisterD) ConnectHandlerD; - -alias int function(Dispatcher) IncomingHandlerF; -alias int function(Dispatcher) OutgoingHandlerF; -alias int function(Dispatcher, RegisterD) ErrorHandlerF; -alias int function(Dispatcher) DisconnectHandlerF; -alias int function(Conduit, RegisterD) ConnectHandlerF; - - -/****************************************************************************** - Dispatcher object. To be used by the SelectLoop to manage callbacks - for events. It may also be used to buffer data inbetween requests. - These can be populated passed to a SelectLoop directly by the end user, - or may be managed by a chosen Protocol. -******************************************************************************/ class Dispatcher { - public - enum State { init, connected, listening, idle, closing }; - - /************************************************************************** - - Standard Ctor, takes a transport_ - - **************************************************************************/ - this (Conduit trans, bool listener = false) +public + this (Conduit trans) { - transport_ = trans; + cond = trans; ibuf_len = 0; - i_offset = 0; o_offset = 0; out_buffers = new CircularSeq!(char[]); log = Log.lookup("dreactor.core.Dispatcher"); } - /********************************************************************** - - Setters for the handlers. These are set by the Protocols as well - - **********************************************************************/ - - void setIncomingHandler(IncomingHandlerD hand) - { - inD = hand; - inF = null; - } - - void setIncomingHandler(IncomingHandlerF hand) - { - inF = hand; - inD = null; - } - - void setOutgoingHandler(OutgoingHandlerD hand) - { - outD = hand; - outF = null; - } + /************************************************************************** - void setOutgoingHandler(OutgoingHandlerF hand) - { - outF = hand; - outD = null; - } - - void setErrorHandler(ErrorHandlerD hand) - { - errD = hand; - errF = null; - } + onSend -- Send method + Called by the vat in response to a FD writeable event. + Sends data, returns amount sent. Unregisters Handler for sending + if there is no more data left to send. - void setErrorHandler(ErrorHandlerF hand) - { - errF = hand; - errD = null; - } - - void setDisconnectHandler(DisconnectHandlerD hand) - { - disD = hand; - disF = null; - } - - void setDisconnectHandler(DisconnectHandlerF hand) - { - disF = hand; - disD = null; - } - - void setConnectHandler(ConnectHandlerD hand) + ***************************************************************************/ + public int onSend() { - conD = hand; - conF = null; - } - - void setConnectHandler(ConnectHandlerF hand) - { - conF = hand; - conD = null; - } - - /********************************************************************** - - Handlers to be called by the SelectLoop when events occur - - **********************************************************************/ - int handleIncoming() - { - if (inD !is null) - return inD(this); - else if (inF !is null) - return inF(this); - else - throw new Exception("no Incoming handler set"); - } - - int handleOutgoing() - { - if (outD !is null) - return outD(this); - else if (outF !is null) - return outF(this); - else - throw new Exception("no Outgoing handler set"); - } - - int handleError(RegisterD reg) - { - if (errD !is null) - return errD(this, reg); - else if (errF !is null) - return errF(this, reg); - } - - int handleDisconnect() - { - if (disD !is null) - return disD(this); - else if (disF !is null) - return disF(this); - } - - int handleConnection(Conduit cond, RegisterD reg ) - { - if (conD !is null) + Logger log = Log.lookup("Handlers.onSend"); + + char[] outbuf = nextBuffer(); + if (outbuf !is null) { - return conD(cond, reg); + int sent = cond.write(outbuf); + if (sent > 0) + { + if (! addOffset(sent)) + { + return UNREGISTER; + } + } + else if (sent == 0) + { + log.error("Select said socket was writable, but sent 0 bytes"); + } + else + { + log.error("Socket send return ERR {}", sent); + } + return REMAIN; } - else if (conF !is null) + else { - return conF(cond, reg); + return UNREGISTER; } } /************************************************************************** - - Sending / Receiving helpers - - **************************************************************************/ - - /************************************************************************** appendOutBuffer @@ -184,7 +69,7 @@ SelectLoop. If it returns false, it was probably already registered. **************************************************************************/ - synchronized bool appendOutBuffer(char[] outbuf) + bool appendOutBuffer(char[] outbuf) { out_buffers.append(outbuf); out_buffers_len++; @@ -193,7 +78,7 @@ else return false; } - + /************************************************************************** addOffset @@ -204,7 +89,7 @@ Returns: false if there is nothing left to send, true if there is. **************************************************************************/ - synchronized bool addOffset(int off) + bool addOffset(int off) in { assert(out_buffers_len > 0); @@ -223,7 +108,7 @@ o_offset += off; return true; } - + /************************************************************************** char[] nextBuffer @@ -242,146 +127,10 @@ return out_buffers.head()[o_offset .. $]; } - /************************************************************************** - - listen - Enable listening on the socket attached to this connectionhandler - - **************************************************************************/ - int listen(IPv4Address addr) - { - (cast(AsyncSocketConduit)transport_).bind(addr).listen(); - state_ = State.listening; - return 0; - } - - Conduit transport() - { - return transport_; - } - /************************************************************************** - - Configuration functions - - **************************************************************************/ - Event events() - { - return events_; - } - void events(Event e) - { - events_ = e; - } - void addEvent(Event e) - { - events_ |= e; - } - void remEvent(Event e) - { - events_ &= ~e; - } - - State getState() {return state_;} - - /* - connection handlers are left out of this because - this method is used by the listener socket to pass - on its handlers to the accepted socket. An accepted - socket will generally do different things onConnection - */ - void setHandlers(Dispatcher other) - { - inD = other.inD; - outD = other.outD; - errD = other.errD; - disD = other.disD; - inF = other.inF; - outF = other.outF; - errF = other.errF; - disF = other.disF; - } - - /************************************************************************** - - Freelist allocators and deallocators - - **************************************************************************/ - static synchronized Dispatcher New(Conduit tran, Dispatcher other = null) - { - Dispatcher hand; - if (freelist) - { - hand = freelist; - freelist = hand.next; - hand.transport_ = tran; - } - else - hand = new Dispatcher(tran); - - if (!(other is null)) - { - hand.setHandlers(other); - } - return hand; - } - - static synchronized void Delete(Dispatcher hand) - { - hand.next = freelist; - freelist = hand.initialize(); - } - -private - - char[] in_buffer; + Conduit cond; CircularSeq!(char[]) out_buffers; int out_buffers_len; int ibuf_len; - int i_offset; int o_offset; Logger log; - - package Conduit transport_; - State state_; - Event events_; - IncomingHandlerD inD; - OutgoingHandlerD outD; - ErrorHandlerD errD; - DisconnectHandlerD disD; - ConnectHandlerD conD; - - IncomingHandlerF inF; - OutgoingHandlerF outF; - ErrorHandlerF errF; - DisconnectHandlerF disF; - ConnectHandlerF conF; - - static Dispatcher freelist; - Dispatcher next; - - /************************************************************************** - Copy ctor, creates a new Dispatcher using the settings - of an existing handler. - **************************************************************************/ - Dispatcher initialize() - { - transport_ = null; - state_ = State.init; - ibuf_len = 0; - i_offset = 0; - o_offset = 0; - out_buffers.clear(); - inD = null; - outD = null; - errD = null; - disD = null; - conD = null; - inF = null; - outF = null; - errF = null; - disF = null; - conF = null; - return this; - } } -
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/dreactor/core/Task.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,84 @@ +module dreactor.core.Task; + +import tango.core.Thread; +import dreactor.core.Vat; +import dreactor.protocol.Protocol; +import dreactor.protocol.Dispatcher; + +alias CircularSeq!(Message) Mailbox; + +class Task +{ +private + Fiber fiber; + Mailbox mailbox; + int id; + Vat vat; + dispatcher[Conduit] dispatchers; + +public + this() + { + fiber = new Fiber(&run); + mailbox = new Mailbox; + } + + void setId(int i) + { + id = i; + } + + Mailbox getMailbox() + { + return mailbox; + } + + void setVat(Vat v) + { + vat = v; + } + + abstract void run(); + +protected + + /*************************************************************************** + receive + User-called function to get the next pending message in the mailbox. + If there are no pending messages, this will yield control back to + the scheduler/vat. + ***************************************************************************/ + + Message receive() + { + Message m = mailbox.head(); + mailbox.removeHead(); + return m; + } + + int getId() { return id;} + + /************************************************************************** + + send + User-called function to send data to the counterpart at the other + end of the connection. This sets up a dispatcher to send + data as the conduit becomes free. + + **************************************************************************/ + int send(char[] outbuf, Conduit c) + { + Dispatcher dis; + if ( ! (dis = (c in dispatchers))) + dis = new Dispatcher(c); + + if (dis.appendOutBuffer(outbuf)) + { + if (!vat.addConnection(dis)) + { + log.error("unable to register mgr"); + } + } + return 0; + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/dreactor/core/Vat.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,200 @@ +/******************************************************************************* + + copyright: Copyright (c) 2008 Rick Richardson. All rights reserved + + license: BSD style: $(LICENSE) + + version: Initial release v0.1 : May 2008 + + author: Rick Richardson + +*******************************************************************************/ + +module dreactor.core.Vat; + +import tango.io.selector.Selector; +import tango.io.selector.model.ISelector; +import tango.core.Exception; +import tango.core.Thread; +import tango.core.Atomic; +import tango.util.collection.CircularSeq; +import tango.util.log.Log; + +import dreactor.transport.AsyncSocketConduit; +import dreactor.core.Task; +import dreactor.util.ThreadSafeQueue; + +static char[] version_string = "Vat.d 0.1 2008-05-31"; + +Logger log; + +enum : int {CLOSE = -2, UNREGISTER = -1, REMAIN = 0, REGISTER = 1, REREGISTER = 2}; +alias Message delegate (Conduit c) HandlerDG; +alias Message function (Conduit c) HandlerFN; + +class TaskAttachment +{ +public + Task task; + HandlerDG dg; + HandlerFN fn; + + this(Task ta, HandlerDG d) + { TaskAttachment t; t.task = ta; t.dg = d; return t; } + + this(Task ta, HandlerFN f) + { TaskAttachment t; t.task = ta; t.fn = f; return t; } + + public Message opCall(Conduit c) { dg is null ? return fn() : return dg(c); } +} + +class Vat +{ +private + Thread thread; + bool running; + + Task[int] tasks; + int taskCount; + +public + + this(Task t) + { + addTask(t); + this(); + } + + this() + { + log = Log.lookup("dreactor.core.Vat"); + + running = true; + thread = new Thread(&eventLoop); + thread.start(); + } + + void addTask(Task t) + { + t.setVat(this); + ++taskCount; + tasks[taskCount] = t; + t.setId(taskCount); + } + + void exit() + { + running = false; + } + + void wait() + { + thread.join(); + } + + bool addConnection() + { + log.trace("adding handler"); + return selector.register(h.transport, h.events(), h); + } + + bool remConnection(Dispatcher handler) + { + return selector.unregister(h.transport); + } + +private + void eventLoop() + { + auto selector = new Selector(); + selector.open(); + do + { + execTasks(); + auto eventCount = selector.select(0.01); + + if (eventCount > 0) + { + // process events + foreach (SelectionKey key; selector.selectedSet()) + { + if (key.isReadable()) + { + // incoming data + log.trace("Read event fired"); + auto conn = cast(Dispatcher) key.attachment; + if ( Dispatcher.State.listening == conn.getState() ) + conn.handleConnection(conn.transport, &addConnection); + else + processReturn(conn.handleIncoming(), selector, conn); + } + else if (key.isWritable()) + { + log.trace("Write event fired"); + auto conn = cast(Dispatcher) key.attachment; + processReturn(conn.handleOutgoing(), selector, conn); + } + else if (key.isHangup()) + { + log.trace("Hangup event fired"); + auto conn = cast(Dispatcher) key.attachment; + processReturn(conn.handleDisconnect(), selector, conn); + } + else if (key.isError() || key.isInvalidHandle()) + { + log.trace("Error event fired"); + // error, close connection + auto conn = cast(Dispatcher) key.attachment; + conn.handleError(&remConnection); + } + } + } + else if (eventCount == 0) + { + /* can't think of anything useful to do here. */ + } + else + { + log.error("Selector.select returned {}", eventCount); + } + + } while (running) + + } + + void execTasks() + { + foreach(int k; tasks.keys) + { + if (tasks[k].state() == Fiber.State.HOLD) + tasks[k].call(); + if (tasks[k].state() == Fiber.State.TERM) + tasks.remove(k); + } + } + + void processReturn(int result, Selector s, Dispatcher h) + { + switch(result) + { + case CLOSE: + s.unregister(h.transport); + h.transport.detach(); + break; + case UNREGISTER: + s.unregister(h.transport); + break; + case REMAIN: + //this space intentially left blank + break; + case REGISTER: + s.register(h.transport, h.events(), h); + break; + case REREGISTER: + s.register(h.transport, h.events(), h); + break; + default: + log.error("processReturn: unknown return value"); + } + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/dreactor/protocol/Protocol.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,26 @@ +module dreactor.protocol.IProtocol; + +struct Message +{ +public + int type; + int errorcode; + Object payload; + static Message opCall(Object buf, int t, int e) + { + Message m; + m.type = t; + errorcode = e; + m.payload = buf; + return m; + } +} + +interface IProtocol +{ + Message handleRead(Conduit c); + Message handleWrite(Conduit c); + Message handleError(Conduit c); + Message handleConnect(Conduit c); + Message handleDisconnect(Conduit c); +}
--- a/dreactor/protocol/RawTcp.d Fri Aug 01 16:30:45 2008 -0400 +++ b/dreactor/protocol/RawTcp.d Tue Aug 12 16:59:56 2008 -0400 @@ -19,29 +19,19 @@ class RawTCPListener { public - this(Dispatcher mgr, Vat sel, IPv4Address addr) + + this(AsyncSocketConduit cond) { - manager = mgr; - mgr.events(Event.Read); - mgr.setOutgoingHandler(&RawTCPHandler.onSend); - mgr.setIncomingHandler(&onReceive); - mgr.setConnectHandler(&accept); - mgr.setErrorHandler(&RawTCPHandler.onError); - mgr.setDisconnectHandler(&RawTCPHandler.onHangup); - mgr.listen(addr); - - sel.addConnection(mgr); - vat = sel; log = Log.lookup("dreactor.protocol.RawTcpServer"); log.info("log initialized"); children = new CircularSeq!(Dispatcher); } + this(Vat sel, IPv4Address addr) { AsyncSocketConduit cond = new AsyncSocketConduit; cond.socket().setAddressReuse(true); - Dispatcher lh = new Dispatcher(cond, true); - this(lh, sel, addr); + this(cond); } ~this() @@ -49,25 +39,22 @@ close(); } - int accept(Conduit cond, RegisterD reg) + AsyncSocketConduit accept(Conduit cond, RegisterD reg) { AsyncSocketConduit newcond = new AsyncSocketConduit; (cast(AsyncSocketConduit)cond).socket().accept(newcond.socket); - Dispatcher h = Dispatcher.New(newcond, manager); h.events(Event.Read); vat.addConnection(h); children.append(h); log.info("accepted new connection"); - return 0; + return newcond; } - int broadcast(char[] outbuf, Dispatcher[] excluded = null) + int broadcast(char[] outbuf, AsyncSocketConduit[] recips) { - foreach(Dispatcher h; children) + foreach(AsyncSocketConduit c; recips) { - if (excluded && excluded.includes(h)) - continue; - if (h.appendOutBuffer(outbuf)) + if (c.appendOutBuffer(outbuf)) { h.addEvent(Event.Write); vat.addConnection(h); @@ -263,6 +250,7 @@ { dataHandler = h; } + private void delegate(char[], Dispatcher) dataHandler; Dispatcher manager;
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/dreactor/protocol/http11_parser.d Tue Aug 12 16:59:56 2008 -0400 @@ -0,0 +1,492 @@ +module dreactor.protocol.http11_parser; + +//#line 1 "http11_parser.rl" +/** + * Copyright (c) 2005 Zed A. Shaw, + * Modified for D by Rick Richardson, 2008 + * You can redistribute it and/or modify it under the same terms as Ruby. + */ + + +alias void delegate (Object data, char[] buf) element_cb; +alias void delegate (Object data, char[] field, char[] value) field_cb; + +class Http11Parser +{ +private + int cs; + size_t body_start; + int content_len; + size_t nread; + size_t mark; + size_t field_start; + size_t field_len; + size_t query_start; + + Object data; + + field_cb http_field; + element_cb request_method; + element_cb request_uri; + element_cb fragment; + element_cb request_path; + element_cb query_string; + element_cb http_version; + element_cb header_done; + +/* + * capitalizes all lower-case ASCII characters, + * converts dashes to underscores. + */ +private void snake_upcase_char(char *c) +{ + if (*c >= 'a' && *c <= 'z') + *c &= ~0x20; + else if (*c == '-') + *c = '_'; +} + +//#define LEN(AT, FPC) (FPC - buffer - parser->AT) +private size_t LEN(char[] buffer, size_t at, char* fpc) +{ + return (fpc - buffer.ptr - at); +} + +//#define MARK(M,FPC) (parser->M = (FPC) - buffer) +private void MARK(char[] buffer, size_t* item, char* fpc) +{ + *item = fpc - buffer.ptr; +} + +/** Machine **/ + +//#line 126 "http11_parser.rl" + + +/** Data **/ + +//#line 70 "http11_parser.d" +static byte[] _http_parser_actions = [ + 0, 1, 0, 1, 2, 1, 3, 1, + 4, 1, 5, 1, 6, 1, 7, 1, + 8, 1, 9, 1, 11, 1, 12, 1, + 13, 2, 1, 2, 2, 10, 7, 2, + 12, 7, 3, 9, 10, 7 +]; + +static short[] _http_parser_key_offsets = [ + 0, 0, 8, 17, 27, 29, 30, 31, + 32, 33, 34, 36, 39, 41, 44, 45, + 61, 62, 78, 80, 81, 90, 99, 105, + 111, 121, 130, 136, 142, 153, 159, 165, + 175, 181, 187, 196, 205, 211, 217, 226, + 235, 244, 253, 262, 271, 280, 289, 298, + 307, 316, 325, 334, 343, 352, 361, 370, + 379, 380 +]; + +static char[] _http_parser_trans_keys = [ + 36u, 95u, 45u, 46u, 48u, 57u, 65u, 90u, + 32u, 36u, 95u, 45u, 46u, 48u, 57u, 65u, + 90u, 42u, 43u, 47u, 58u, 45u, 57u, 65u, + 90u, 97u, 122u, 32u, 35u, 72u, 84u, 84u, + 80u, 47u, 48u, 57u, 46u, 48u, 57u, 48u, + 57u, 13u, 48u, 57u, 10u, 13u, 33u, 124u, + 126u, 35u, 39u, 42u, 43u, 45u, 46u, 48u, + 57u, 65u, 90u, 94u, 122u, 10u, 33u, 58u, + 124u, 126u, 35u, 39u, 42u, 43u, 45u, 46u, + 48u, 57u, 65u, 90u, 94u, 122u, 13u, 32u, + 13u, 32u, 37u, 60u, 62u, 127u, 0u, 31u, + 34u, 35u, 32u, 37u, 60u, 62u, 127u, 0u, + 31u, 34u, 35u, 48u, 57u, 65u, 70u, 97u, + 102u, 48u, 57u, 65u, 70u, 97u, 102u, 43u, + 58u, 45u, 46u, 48u, 57u, 65u, 90u, 97u, + 122u, 32u, 34u, 35u, 37u, 60u, 62u, 127u, + 0u, 31u, 48u, 57u, 65u, 70u, 97u, 102u, + 48u, 57u, 65u, 70u, 97u, 102u, 32u, 34u, + 35u, 37u, 59u, 60u, 62u, 63u, 127u, 0u, + 31u, 48u, 57u, 65u, 70u, 97u, 102u, 48u, + 57u, 65u, 70u, 97u, 102u, 32u, 34u, 35u, + 37u, 60u, 62u, 63u, 127u, 0u, 31u, 48u, + 57u, 65u, 70u, 97u, 102u, 48u, 57u, 65u, + 70u, 97u, 102u, 32u, 34u, 35u, 37u, 60u, + 62u, 127u, 0u, 31u, 32u, 34u, 35u, 37u, + 60u, 62u, 127u, 0u, 31u, 48u, 57u, 65u, + 70u, 97u, 102u, 48u, 57u, 65u, 70u, 97u, + 102u, 32u, 36u, 95u, 45u, 46u, 48u, 57u, + 65u, 90u, 32u, 36u, 95u, 45u, 46u, 48u, + 57u, 65u, 90u, 32u, 36u, 95u, 45u, 46u, + 48u, 57u, 65u, 90u, 32u, 36u, 95u, 45u, + 46u, 48u, 57u, 65u, 90u, 32u, 36u, 95u, + 45u, 46u, 48u, 57u, 65u, 90u, 32u, 36u, + 95u, 45u, 46u, 48u, 57u, 65u, 90u, 32u, + 36u, 95u, 45u, 46u, 48u, 57u, 65u, 90u, + 32u, 36u, 95u, 45u, 46u, 48u, 57u, 65u, + 90u, 32u, 36u, 95u, 45u, 46u, 48u, 57u, + 65u, 90u, 32u, 36u, 95u, 45u, 46u, 48u, + 57u, 65u, 90u, 32u, 36u, 95u, 45u, 46u, + 48u, 57u, 65u, 90u, 32u, 36u, 95u, 45u, + 46u, 48u, 57u, 65u, 90u, 32u, 36u, 95u, + 45u, 46u, 48u, 57u, 65u, 90u, 32u, 36u, + 95u, 45u, 46u, 48u, 57u, 65u, 90u, 32u, + 36u, 95u, 45u, 46u, 48u, 57u, 65u, 90u, + 32u, 36u, 95u, 45u, 46u, 48u, 57u, 65u, + 90u, 32u, 36u, 95u, 45u, 46u, 48u, 57u, + 65u, 90u, 32u, 36u, 95u, 45u, 46u, 48u, + 57u, 65u, 90u, 32u, 0 +]; + +static byte[] _http_parser_single_lengths = [ + 0, 2, 3, 4, 2, 1, 1, 1, + 1, 1, 0, 1, 0, 1, 1, 4, + 1, 4, 2, 1, 5, 5, 0, 0, + 2, 7, 0, 0, 9, 0, 0, 8, + 0, 0, 7, 7, 0, 0, 3, 3, + 3, 3, 3, 3, 3, 3, 3, 3, + 3, 3, 3, 3, 3, 3, 3, 3, + 1, 0 +]; + +static byte[] _http_parser_range_lengths = [ + 0, 3, 3, 3, 0, 0, 0, 0, + 0, 0, 1, 1, 1, 1, 0, 6, + 0, 6, 0, 0, 2, 2, 3, 3, + 4, 1, 3, 3, 1, 3, 3, 1, + 3, 3, 1, 1, 3, 3, 3, 3, + 3, 3, 3, 3, 3, 3, 3, 3, + 3, 3, 3, 3, 3, 3, 3, 3, + 0, 0 +]; + +static short[] _http_parser_index_offsets = [ + 0, 0, 6, 13, 21, 24, 26, 28, + 30, 32, 34, 36, 39, 41, 44, 46, + 57, 59, 70, 73, 75, 83, 91, 95, + 99, 106, 115, 119, 123, 134, 138, 142, + 152, 156, 160, 169, 178, 182, 186, 193, + 200, 207, 214, 221, 228, 235, 242, 249, + 256, 263, 270, 277, 284, 291, 298, 305, + 312, 314 +]; + +static byte[] _http_parser_indicies = [ + 0, 0, 0, 0, 0, 1, 2, 3, + 3, 3, 3, 3, 1, 4, 5, 6, + 7, 5, 5, 5, 1, 8, 9, 1, + 10, 1, 11, 1, 12, 1, 13, 1, + 14, 1, 15, 1, 16, 15, 1, 17, + 1, 18, 17, 1, 19, 1, 20, 21, + 21, 21, 21, 21, 21, 21, 21, 21, + 1, 22, 1, 23, 24, 23, 23, 23, + 23, 23, 23, 23, 23, 1, 26, 27, + 25, 26, 28, 29, 31, 1, 1, 1, + 1, 1, 30, 29, 33, 1, 1, 1, + 1, 1, 32, 34, 34, 34, 1, 32, + 32, 32, 1, 35, 36, 35, 35, 35, + 35, 1, 8, 1, 9, 37, 1, 1, + 1, 1, 36, 38, 38, 38, 1, 36, + 36, 36, 1, 39, 1, 41, 42, 43, + 1, 1, 44, 1, 1, 40, 45, 45, + 45, 1, 40, 40, 40, 1, 8, 1, + 9, 47, 1, 1, 48, 1, 1, 46, + 49, 49, 49, 1, 46, 46, 46, 1, + 50, 1, 52, 53, 1, 1, 1, 1, + 51, 54, 1, 56, 57, 1, 1, 1, + 1, 55, 58, 58, 58, 1, 55, 55, + 55, 1, 2, 59, 59, 59, 59, 59, + 1, 2, 60, 60, 60, 60, 60, 1, + 2, 61, 61, 61, 61, 61, 1, 2, + 62, 62, 62, 62, 62, 1, 2, 63, + 63, 63, 63, 63, 1, 2, 64, 64, + 64, 64, 64, 1, 2, 65, 65, 65, + 65, 65, 1, 2, 66, 66, 66, 66, + 66, 1, 2, 67, 67, 67, 67, 67, + 1, 2, 68, 68, 68, 68, 68, 1, + 2, 69, 69, 69, 69, 69, 1, 2, + 70, 70, 70, 70, 70, 1, 2, 71, + 71, 71, 71, 71, 1, 2, 72, 72, + 72, 72, 72, 1, 2, 73, 73, 73, + 73, 73, 1, 2, 74, 74, 74, 74, + 74, 1, 2, 75, 75, 75, 75, 75, + 1, 2, 76, 76, 76, 76, 76, 1, + 2, 1, 1, 0 +]; + +static byte[] _http_parser_trans_targs_wi = [ + 2, 0, 3, 38, 4, 24, 28, 25, + 5, 20, 6, 7, 8, 9, 10, 11, + 12, 13, 14, 15, 16, 17, 57, 17, + 18, 19, 14, 18, 19, 5, 21, 22, + 21, 22, 23, 24, 25, 26, 27, 5, + 28, 20, 29, 31, 34, 30, 31, 32, + 34, 33, 5, 35, 20, 36, 5, 35, + 20, 36, 37, 39, 40, 41, 42, 43, + 44, 45, 46, 47, 48, 49, 50, 51, + 52, 53, 54, 55, 56 +]; + +static byte[] _http_parser_trans_actions_wi = [ + 1, 0, 11, 0, 1, 1, 1, 1, + 13, 13, 1, 0, 0, 0, 0, 0, + 0, 0, 19, 0, 0, 25, 23, 3, + 5, 7, 9, 7, 0, 15, 1, 1, + 0, 0, 0, 0, 0, 0, 0, 31, + 0, 31, 0, 21, 21, 0, 0, 0, + 0, 0, 34, 17, 34, 17, 28, 0, + 28, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0 +]; + +static int http_parser_start = 1; +static int http_parser_first_final = 57; +static int http_parser_error = 0; + +static int http_parser_en_main = 1; + +//#line 130 "http11_parser.rl" + +this () +{ + cs = 0; + +//#line 255 "http11_parser.d" + { + cs = http_parser_start; + } +//#line 135 "http11_parser.rl" + + body_start = 0; + content_len = 0; + mark = 0; + nread = 0; + field_len = 0; + field_start = 0; +} + + +/** exec **/ +public size_t execute(char[] buffer, size_t off = 0) { + char* p, pe; + int len = buffer.length; + + assert(off <= len && "offset past end of buffer"); + + p = buffer.ptr+off; + pe = buffer.ptr+len; + + /* assert(*pe == '\0' && "pointer does not end on NUL"); */ + assert(pe - p == len - off && "pointers aren't same distance"); + + +//#line 284 "http11_parser.d" + { + int _klen; + uint _trans; + byte* _acts; + uint _nacts; + char* _keys; + + if ( p == pe ) + goto _out; + if ( cs == 0 ) + goto _out; +_resume: + _keys = &_http_parser_trans_keys[_http_parser_key_offsets[cs]]; + _trans = _http_parser_index_offsets[cs]; + + _klen = _http_parser_single_lengths[cs]; + if ( _klen > 0 ) { + char* _lower = _keys; + char* _mid; + char* _upper = _keys + _klen - 1; + while (1) { + if ( _upper < _lower ) + break; + + _mid = _lower + ((_upper-_lower) >> 1); + if ( (*p) < *_mid ) + _upper = _mid - 1; + else if ( (*p) > *_mid ) + _lower = _mid + 1; + else { + _trans += (_mid - _keys); + goto _match; + } + } + _keys += _klen; + _trans += _klen; + } + + _klen = _http_parser_range_lengths[cs]; + if ( _klen > 0 ) { + char* _lower = _keys; + char* _mid; + char* _upper = _keys + (_klen<<1) - 2; + while (1) { + if ( _upper < _lower ) + break; + + _mid = _lower + (((_upper-_lower) >> 1) & ~1); + if ( (*p) < _mid[0] ) + _upper = _mid - 2; + else if ( (*p) > _mid[1] ) + _lower = _mid + 2; + else { + _trans += ((_mid - _keys)>>1); + goto _match; + } + } + _trans += _klen; + } + +_match: + _trans = _http_parser_indicies[_trans]; + cs = _http_parser_trans_targs_wi[_trans]; + + if ( _http_parser_trans_actions_wi[_trans] == 0 ) + goto _again; + + _acts = &_http_parser_actions[_http_parser_trans_actions_wi[_trans]]; + _nacts = cast(uint) *_acts++; + while ( _nacts-- > 0 ) + { + switch ( *_acts++ ) + { + case 0: +//#line 67 "http11_parser.rl" + {buffer.MARK(&mark, p); } + break; + case 1: +//#line 70 "http11_parser.rl" + { buffer.MARK(&field_start, p); } + break; + case 2: +//#line 71 "http11_parser.rl" + { snake_upcase_char(cast(char *)p); } + break; + case 3: +//#line 72 "http11_parser.rl" + { + field_len = buffer.LEN(field_start, p); + } + break; + case 4: +//#line 76 "http11_parser.rl" + { buffer.MARK(&mark, p); } + break; + case 5: +//#line 77 "http11_parser.rl" + { + if(http_field) + http_field(data, + buffer[field_start .. field_start+field_len], + buffer[mark .. $]); + } + break; + case 6: +//#line 85 "http11_parser.rl" + { + if(request_method) + request_method(data, buffer[mark .. $]); + } + break; + case 7: +//#line 90 "http11_parser.rl" + { + if(request_uri) + request_uri(data, buffer[mark .. $]); + } + break; + case 8: +//#line 95 "http11_parser.rl" + { + if(fragment) + fragment(data, buffer[mark .. $]); + } + break; + case 9: +//#line 100 "http11_parser.rl" + {buffer.MARK(&query_start, p); } + break; + case 10: +//#line 102 "http11_parser.rl" + { + if(query_string) + query_string(data, buffer[query_start .. $]); + } + break; + case 11: +//#line 107 "http11_parser.rl" + { + if(http_version) + http_version(data, buffer[mark .. $]); + } + break; + case 12: +//#line 112 "http11_parser.rl" + { + if(request_path) + request_path(data, buffer[mark .. $]); + } + break; + case 13: +//#line 117 "http11_parser.rl" + { + body_start = p - buffer.ptr + 1; + if(header_done) + header_done(data, p[1 .. (pe - p - 1)]); + if (true) goto _out; + } + break; +//#line 446 "http11_parser.d" + default: break; + } + } + +_again: + if ( cs == 0 ) + goto _out; + if ( ++p != pe ) + goto _resume; + _out: {} + } +//#line 159 "http11_parser.rl" + + //parser->cs = cs; + nread += p - (buffer.ptr + off); + + assert(p <= pe && "buffer overflow after parsing execute"); + assert(nread <= len && "nread longer than length"); + assert(body_start <= len && "body starts after buffer end"); + assert(mark < len && "mark is after buffer end"); + assert(field_len <= len && "field has length longer than whole buffer"); + assert(field_start < len && "field starts after buffer end"); + + return(nread); +} + +public int finish() +{ + if (has_error() ) { + return -1; + } else if (is_finished() ) { + return 1; + } else { + return 0; + } +} + +public bool has_error() { + return cs == http_parser_error; +} + +public bool is_finished() { + return cs >= http_parser_first_final; +} + +public void setData(Object d) +{ + data = d; +} +}