X-Git-Url: http://nitlanguage.org diff --git a/lib/libevent.nit b/lib/libevent.nit index ff9a9f5..c01d417 100644 --- a/lib/libevent.nit +++ b/lib/libevent.nit @@ -42,44 +42,21 @@ in "C" `{ // Callback forwarded to 'Connection.read_callback_native' static void c_read_cb(struct bufferevent *bev, Connection ctx) { - // TODO move to Nit code - struct evbuffer *input = bufferevent_get_input(bev); - size_t len = evbuffer_get_length(input); - char* cstr = malloc(len); - evbuffer_remove(input, cstr, len); - Connection_read_callback_native(ctx, cstr, len); + Connection_read_callback_native(ctx, bev); } // Callback forwarded to 'Connection.event_callback' static void c_event_cb(struct bufferevent *bev, short events, Connection ctx) { - Connection_event_callback(ctx, events); - - // TODO move to Nit code - if (events & BEV_EVENT_ERROR) - perror("Error from bufferevent"); - if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) { - bufferevent_free(bev); - Connection_decr_ref(ctx); - } + int release = Connection_event_callback(ctx, events); + if (release) Connection_decr_ref(ctx); } - // Callback fowarded to 'ConnectionFactory.spawn_connection' - static void accept_conn_cb(struct evconnlistener *listener, evutil_socket_t fd, + // Callback forwarded to 'ConnectionFactory.accept_connection' + static void accept_connection_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *address, int socklen, ConnectionFactory ctx) { - // TODO move to Nit code - struct event_base *base = evconnlistener_get_base(listener); - struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE); - - Connection nit_con = ConnectionFactory_spawn_connection(ctx, bev); - Connection_incr_ref(nit_con); - - bufferevent_setcb(bev, - (bufferevent_data_cb)c_read_cb, - (bufferevent_data_cb)c_write_cb, - (bufferevent_event_cb)c_event_cb, nit_con); - bufferevent_enable(bev, EV_READ|EV_WRITE); + ConnectionFactory_accept_connection(ctx, listener, fd, address, socklen); } `} @@ -100,21 +77,23 @@ extern class NativeEventBase `{ struct event_base * `} # # This loop will run the event base until either there are no more added # events, or until something calls `exit_loop`. - fun dispatch `{ event_base_dispatch(recv); `} + fun dispatch `{ event_base_dispatch(self); `} # Exit the event loop # # TODO support timer - fun exit_loop `{ event_base_loopexit(recv, NULL); `} + fun exit_loop `{ event_base_loopexit(self, NULL); `} # Destroy this instance - fun destroy `{ event_base_free(recv); `} + fun destroy `{ event_base_free(self); `} end # Spawned to manage a specific connection # # TODO, use polls class Connection + super Writer + # Closing this connection has been requested, but may not yet be `closed` var close_requested = false @@ -125,8 +104,9 @@ class Connection var native_buffer_event: NativeBufferEvent # Close this connection if possible, otherwise mark it to be closed later - fun close + redef fun close do + if closed then return var success = native_buffer_event.destroy close_requested = true closed = success @@ -135,76 +115,205 @@ class Connection # Callback method on a write event fun write_callback do - if close_requested and not closed then close + if close_requested then close end - private fun read_callback_native(cstr: NativeString, len: Int) + private fun read_callback_native(bev: NativeBufferEvent) do - read_callback(cstr.to_s_with_length(len)) + var evbuffer = bev.input_buffer + var len = evbuffer.length + var buf = new NativeString(len) + evbuffer.remove(buf, len) + var str = buf.to_s_with_length(len) + read_callback str end # Callback method when data is available to read fun read_callback(content: String) do - if close_requested and not closed then close + if close_requested then close end - # Callback method on events - fun event_callback(events: Int) do end + # Callback method on events: EOF, user-defined timeout and unrecoverable errors + # + # Returns `true` if the native handles to `self` can be released. + fun event_callback(events: Int): Bool + do + if events & bev_event_error != 0 or events & bev_event_eof != 0 then + if events & bev_event_error != 0 then print_error "Error from bufferevent" + close + return true + end + + return false + end # Write a string to the connection - fun write(str: String) + redef fun write(str) do - native_buffer_event.write(str.to_cstring, str.length) + if close_requested then return + native_buffer_event.write(str.to_cstring, str.bytelen) + end + + redef fun write_byte(byte) + do + if close_requested then return + native_buffer_event.write_byte(byte) + end + + redef fun write_bytes(bytes) + do + if close_requested then return + native_buffer_event.write(bytes.items, bytes.length) end # Write a file to the connection # - # require: `path.file_exists` + # If `not path.file_exists`, the method returns. fun write_file(path: String) do - assert path.file_exists + if close_requested then return var file = new FileReader.open(path) - var output = native_buffer_event.output_buffer - var fd = file.fd - var length = file.file_stat.size + if file.last_error != null then + var error = new IOError("Failed to open file at '{path}'") + error.cause = file.last_error + self.last_error = error + file.close + return + end - output.add_file(fd, 0, length) + var stat = file.file_stat + if stat == null then + last_error = new IOError("Failed to stat file at '{path}'") + file.close + return + end + + var err = native_buffer_event.output_buffer.add_file(file.fd, 0, stat.size) + if err then + last_error = new IOError("Failed to add file at '{path}'") + file.close + end end end +# --- +# Error code for event callbacks + +# error encountered while reading +fun bev_event_reading: Int `{ return BEV_EVENT_READING; `} + +# error encountered while writing +fun bev_event_writing: Int `{ return BEV_EVENT_WRITING; `} + +# eof file reached +fun bev_event_eof: Int `{ return BEV_EVENT_EOF; `} + +# unrecoverable error encountered +fun bev_event_error: Int `{ return BEV_EVENT_ERROR; `} + +# user-specified timeout reached +fun bev_event_timeout: Int `{ return BEV_EVENT_TIMEOUT; `} + +# connect operation finished. +fun bev_event_connected: Int `{ return BEV_EVENT_CONNECTED; `} + +# --- +# Options that can be specified when creating a `NativeBufferEvent` + +# Close the underlying file descriptor/bufferevent/whatever when this bufferevent is freed. +fun bev_opt_close_on_free: Int `{ return BEV_OPT_CLOSE_ON_FREE; `} + +# If threading is enabled, protect the operations on this bufferevent with a lock. +fun bev_opt_threadsafe: Int `{ return BEV_OPT_THREADSAFE; `} + +# Run callbacks deferred in the event loop. +fun bev_opt_defer_callbacks: Int `{ return BEV_OPT_DEFER_CALLBACKS; `} + +# If set, callbacks are executed without locks being held on the bufferevent. +fun bev_opt_unlock_callbacks: Int `{ return BEV_OPT_UNLOCK_CALLBACKS; `} + +# --- +# Options for `NativeBufferEvent::enable` + +# Read operation +fun ev_read: Int `{ return EV_READ; `} + +# Write operation +fun ev_write: Int `{ return EV_WRITE; `} + +# --- + # A buffer event structure, strongly associated to a connection, an input buffer and an output_buffer extern class NativeBufferEvent `{ struct bufferevent * `} + + # Socket-based `NativeBufferEvent` that reads and writes data onto a network + new socket(base: NativeEventBase, fd, options: Int) `{ + return bufferevent_socket_new(base, fd, options); + `} + + # Enable a bufferevent. + fun enable(operation: Int) `{ + bufferevent_enable(self, operation); + `} + + # Set callbacks to `read_callback_native`, `write_callback` and `event_callback` of `conn` + fun setcb(conn: Connection) import Connection.read_callback_native, + Connection.write_callback, Connection.event_callback, NativeString `{ + Connection_incr_ref(conn); + bufferevent_setcb(self, + (bufferevent_data_cb)c_read_cb, + (bufferevent_data_cb)c_write_cb, + (bufferevent_event_cb)c_event_cb, conn); + `} + # Write `length` bytes of `line` fun write(line: NativeString, length: Int): Int `{ - return bufferevent_write(recv, line, length); + return bufferevent_write(self, line, length); + `} + + # Write the byte `value` + fun write_byte(value: Byte): Int `{ + unsigned char byt = (unsigned char)value; + return bufferevent_write(self, &byt, 1); `} # Check if we have anything left in our buffers. If so, we set our connection to be closed # on a callback. Otherwise we close it and free it right away. fun destroy: Bool `{ - struct evbuffer* out = bufferevent_get_output(recv); - struct evbuffer* in = bufferevent_get_input(recv); + struct evbuffer* out = bufferevent_get_output(self); + struct evbuffer* in = bufferevent_get_input(self); if(evbuffer_get_length(in) > 0 || evbuffer_get_length(out) > 0) { return 0; } else { - bufferevent_free(recv); + bufferevent_free(self); return 1; } `} # The output buffer associated to `self` - fun output_buffer: OutputNativeEvBuffer `{ return bufferevent_get_output(recv); `} + fun output_buffer: OutputNativeEvBuffer `{ return bufferevent_get_output(self); `} # The input buffer associated to `self` - fun input_buffer: InputNativeEvBuffer `{ return bufferevent_get_input(recv); `} + fun input_buffer: InputNativeEvBuffer `{ return bufferevent_get_input(self); `} + + # Read data from this buffer + fun read_buffer(buf: NativeEvBuffer): Int `{ return bufferevent_read_buffer(self, buf); `} + + # Write data to this buffer + fun write_buffer(buf: NativeEvBuffer): Int `{ return bufferevent_write_buffer(self, buf); `} end # A single buffer extern class NativeEvBuffer `{ struct evbuffer * `} # Length of data in this buffer - fun length: Int `{ return evbuffer_get_length(recv); `} + fun length: Int `{ return evbuffer_get_length(self); `} + + # Read data from an evbuffer and drain the bytes read + fun remove(buffer: NativeString, len: Int) `{ + evbuffer_remove(self, buffer, len); + `} end # An input buffer @@ -212,7 +321,7 @@ extern class InputNativeEvBuffer super NativeEvBuffer # Empty/clear `length` data from buffer - fun drain(length: Int) `{ evbuffer_drain(recv, length); `} + fun drain(length: Int) `{ evbuffer_drain(self, length); `} end # An output buffer @@ -221,7 +330,7 @@ extern class OutputNativeEvBuffer # Add file to buffer fun add_file(fd, offset, length: Int): Bool `{ - return evbuffer_add_file(recv, fd, offset, length); + return evbuffer_add_file(self, fd, offset, length); `} end @@ -229,8 +338,7 @@ end extern class ConnectionListener `{ struct evconnlistener * `} private new bind_to(base: NativeEventBase, address: NativeString, port: Int, factory: ConnectionFactory) - import ConnectionFactory.spawn_connection, error_callback, Connection.read_callback_native, - Connection.write_callback, Connection.event_callback `{ + import ConnectionFactory.accept_connection, error_callback `{ struct sockaddr_in sin; struct evconnlistener *listener; @@ -238,13 +346,17 @@ extern class ConnectionListener `{ struct evconnlistener * `} struct hostent *hostent = gethostbyname(address); + if (!hostent) { + return NULL; + } + memset(&sin, 0, sizeof(sin)); sin.sin_family = hostent->h_addrtype; sin.sin_port = htons(port); memcpy( &(sin.sin_addr.s_addr), (const void*)hostent->h_addr, hostent->h_length ); listener = evconnlistener_new_bind(base, - (evconnlistener_cb)accept_conn_cb, factory, + (evconnlistener_cb)accept_connection_cb, factory, LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, -1, (struct sockaddr*)&sin, sizeof(sin)); @@ -256,7 +368,7 @@ extern class ConnectionListener `{ struct evconnlistener * `} `} # Get the `NativeEventBase` associated to `self` - fun base: NativeEventBase `{ return evconnlistener_get_base(recv); `} + fun base: NativeEventBase `{ return evconnlistener_get_base(self); `} # Callback method on listening error fun error_callback do @@ -277,10 +389,22 @@ class ConnectionFactory # The `NativeEventBase` for the dispatch loop of this factory var event_base: NativeEventBase - # On new connection, create the handler `Connection` object - fun spawn_connection(nat_buf_ev: NativeBufferEvent): Connection + # Accept a connection on `listener` + # + # By default, it creates a new NativeBufferEvent and calls `spawn_connection`. + fun accept_connection(listener: ConnectionListener, fd: Int, address: Pointer, socklen: Int) + do + var base = listener.base + var bev = new NativeBufferEvent.socket(base, fd, bev_opt_close_on_free) + var conn = spawn_connection(bev) + bev.enable ev_read|ev_write + bev.setcb conn + end + + # Create a new `Connection` object for `buffer_event` + fun spawn_connection(buffer_event: NativeBufferEvent): Connection do - return new Connection(nat_buf_ev) + return new Connection(buffer_event) end # Listen on `address`:`port` for new connection, which will callback `spawn_connection`