lib/libevent.nit

   1 # This file is part of NIT (http://www.nitlanguage.org).
   2 #
   3 # Copyright 2013 Jean-Philippe Caissy <jpcaissy@piji.ca>
   4 # Copyright 2014 Alexis Laferrière <alexis.laf@xymus.net>
   5 #
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #     http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17
  18 # Low-level wrapper around the libevent library to manage events on file descriptors
  19 #
  20 # For mor information, refer to the libevent documentation at
  21 # http://monkey.org/~provos/libevent/doxygen-2.0.1/
  22 module libevent is pkgconfig("libevent")
  23
  24 in "C header" `{
  25         #include <sys/stat.h>
  26         #include <sys/types.h>
  27         #include <fcntl.h>
  28         #include <errno.h>
  29         #include <sys/socket.h>
  30
  31         #include <event2/listener.h>
  32         #include <event2/bufferevent.h>
  33         #include <event2/buffer.h>
  34 `}
  35
  36 in "C" `{
  37         // Callback forwarded to 'Connection.write_callback'
  38         static void c_write_cb(struct bufferevent *bev, Connection ctx) {
  39                 Connection_write_callback(ctx);
  40         }
  41
  42         // Callback forwarded to 'Connection.read_callback_native'
  43         static void c_read_cb(struct bufferevent *bev, Connection ctx)
  44         {
  45                 Connection_read_callback_native(ctx, bev);
  46         }
  47
  48         // Callback forwarded to 'Connection.event_callback'
  49         static void c_event_cb(struct bufferevent *bev, short events, Connection ctx)
  50         {
  51                 int release = Connection_event_callback(ctx, events);
  52                 if (release) Connection_decr_ref(ctx);
  53         }
  54
  55         // Callback forwarded to 'ConnectionFactory.accept_connection'
  56         static void accept_connection_cb(struct evconnlistener *listener, evutil_socket_t fd,
  57                 struct sockaddr *address, int socklen, ConnectionFactory ctx)
  58         {
  59                 ConnectionFactory_accept_connection(ctx, listener, fd, address, socklen);
  60         }
  61 `}
  62
  63 # Structure to hold information and state for a Libevent dispatch loop.
  64 #
  65 # The event_base lies at the center of Libevent; every application will
  66 # have one.  It keeps track of all pending and active events, and
  67 # notifies your application of the active ones.
  68 extern class NativeEventBase `{ struct event_base * `}
  69
  70         # Create a new event_base to use with the rest of Libevent
  71         new `{ return event_base_new(); `}
  72
  73         # Has `self` been correctly initialized?
  74         fun is_valid: Bool do return not address_is_null
  75
  76         # Event dispatching loop
  77         #
  78         # This loop will run the event base until either there are no more added
  79         # events, or until something calls `exit_loop`.
  80         fun dispatch `{ event_base_dispatch(self); `}
  81
  82         # Exit the event loop
  83         #
  84         # TODO support timer
  85         fun exit_loop `{ event_base_loopexit(self, NULL); `}
  86
  87         # Destroy this instance
  88         fun destroy `{ event_base_free(self); `}
  89 end
  90
  91 # Spawned to manage a specific connection
  92 #
  93 # TODO, use polls
  94 class Connection
  95         super Writer
  96
  97         # Closing this connection has been requested, but may not yet be `closed`
  98         var close_requested = false
  99
 100         # This connection is closed
 101         var closed = false
 102
 103         # The native libevent linked to `self`
 104         var native_buffer_event: NativeBufferEvent
 105
 106         # Close this connection if possible, otherwise mark it to be closed later
 107         redef fun close
 108         do
 109                 if closed then return
 110                 var success = native_buffer_event.destroy
 111                 close_requested = true
 112                 closed = success
 113         end
 114
 115         # Callback method on a write event
 116         fun write_callback
 117         do
 118                 if close_requested then close
 119         end
 120
 121         private fun read_callback_native(bev: NativeBufferEvent)
 122         do
 123                 var evbuffer = bev.input_buffer
 124                 var len = evbuffer.length
 125                 var buf = new NativeString(len)
 126                 evbuffer.remove(buf, len)
 127                 var str = buf.to_s_with_length(len)
 128                 read_callback str
 129         end
 130
 131         # Callback method when data is available to read
 132         fun read_callback(content: String)
 133         do
 134                 if close_requested then close
 135         end
 136
 137         # Callback method on events: EOF, user-defined timeout and unrecoverable errors
 138         #
 139         # Returns `true` if the native handles to `self` can be released.
 140         fun event_callback(events: Int): Bool
 141         do
 142                 if events & bev_event_error != 0 or events & bev_event_eof != 0 then
 143                         if events & bev_event_error != 0 then print_error "Error from bufferevent"
 144                         close
 145                         return true
 146                 end
 147
 148                 return false
 149         end
 150
 151         # Write a string to the connection
 152         redef fun write(str)
 153         do
 154                 if close_requested then return
 155                 native_buffer_event.write(str.to_cstring, str.bytelen)
 156         end
 157
 158         redef fun write_byte(byte)
 159         do
 160                 if close_requested then return
 161                 native_buffer_event.write_byte(byte)
 162         end
 163
 164         redef fun write_bytes(bytes)
 165         do
 166                 if close_requested then return
 167                 native_buffer_event.write(bytes.items, bytes.length)
 168         end
 169
 170         # Write a file to the connection
 171         #
 172         # If `not path.file_exists`, the method returns.
 173         fun write_file(path: String)
 174         do
 175                 if close_requested then return
 176
 177                 var file = new FileReader.open(path)
 178                 if file.last_error != null then
 179                         var error = new IOError("Failed to open file at '{path}'")
 180                         error.cause = file.last_error
 181                         self.last_error = error
 182                         file.close
 183                         return
 184                 end
 185
 186                 var stat = file.file_stat
 187                 if stat == null then
 188                         last_error = new IOError("Failed to stat file at '{path}'")
 189                         file.close
 190                         return
 191                 end
 192
 193                 var err = native_buffer_event.output_buffer.add_file(file.fd, 0, stat.size)
 194                 if err then
 195                         last_error = new IOError("Failed to add file at '{path}'")
 196                         file.close
 197                 end
 198         end
 199 end
 200
 201 # ---
 202 # Error code for event callbacks
 203
 204 # error encountered while reading
 205 fun bev_event_reading: Int `{ return BEV_EVENT_READING; `}
 206
 207 # error encountered while writing
 208 fun bev_event_writing: Int `{ return BEV_EVENT_WRITING; `}
 209
 210 # eof file reached
 211 fun bev_event_eof: Int `{ return BEV_EVENT_EOF; `}
 212
 213 # unrecoverable error encountered
 214 fun bev_event_error: Int `{ return BEV_EVENT_ERROR; `}
 215
 216 # user-specified timeout reached
 217 fun bev_event_timeout: Int `{ return BEV_EVENT_TIMEOUT; `}
 218
 219 # connect operation finished.
 220 fun bev_event_connected: Int `{ return BEV_EVENT_CONNECTED; `}
 221
 222 # ---
 223 # Options that can be specified when creating a `NativeBufferEvent`
 224
 225 # Close the underlying file descriptor/bufferevent/whatever when this bufferevent is freed.
 226 fun bev_opt_close_on_free: Int `{ return BEV_OPT_CLOSE_ON_FREE; `}
 227
 228 # If threading is enabled, protect the operations on this bufferevent with a lock.
 229 fun bev_opt_threadsafe: Int `{ return BEV_OPT_THREADSAFE; `}
 230
 231 # Run callbacks deferred in the event loop.
 232 fun bev_opt_defer_callbacks: Int `{ return BEV_OPT_DEFER_CALLBACKS; `}
 233
 234 # If set, callbacks are executed without locks being held on the bufferevent.
 235 fun bev_opt_unlock_callbacks: Int `{ return BEV_OPT_UNLOCK_CALLBACKS; `}
 236
 237 # ---
 238 # Options for `NativeBufferEvent::enable`
 239
 240 # Read operation
 241 fun ev_read: Int `{ return EV_READ; `}
 242
 243 # Write operation
 244 fun ev_write: Int `{ return EV_WRITE; `}
 245
 246 # ---
 247
 248 # A buffer event structure, strongly associated to a connection, an input buffer and an output_buffer
 249 extern class NativeBufferEvent `{ struct bufferevent * `}
 250
 251         # Socket-based `NativeBufferEvent` that reads and writes data onto a network
 252         new socket(base: NativeEventBase, fd, options: Int) `{
 253                 return bufferevent_socket_new(base, fd, options);
 254         `}
 255
 256         # Enable a bufferevent.
 257         fun enable(operation: Int) `{
 258                 bufferevent_enable(self, operation);
 259         `}
 260
 261         # Set callbacks to `read_callback_native`, `write_callback` and `event_callback` of `conn`
 262         fun setcb(conn: Connection) import Connection.read_callback_native,
 263         Connection.write_callback, Connection.event_callback, NativeString `{
 264                 Connection_incr_ref(conn);
 265                 bufferevent_setcb(self,
 266                         (bufferevent_data_cb)c_read_cb,
 267                         (bufferevent_data_cb)c_write_cb,
 268                         (bufferevent_event_cb)c_event_cb, conn);
 269         `}
 270
 271         # Write `length` bytes of `line`
 272         fun write(line: NativeString, length: Int): Int `{
 273                 return bufferevent_write(self, line, length);
 274         `}
 275
 276         # Write the byte `value`
 277         fun write_byte(value: Byte): Int `{
 278                 unsigned char byt = (unsigned char)value;
 279                 return bufferevent_write(self, &byt, 1);
 280         `}
 281
 282         # Check if we have anything left in our buffers. If so, we set our connection to be closed
 283         # on a callback. Otherwise we close it and free it right away.
 284         fun destroy: Bool `{
 285                 struct evbuffer* out = bufferevent_get_output(self);
 286                 struct evbuffer* in = bufferevent_get_input(self);
 287                 if(evbuffer_get_length(in) > 0 || evbuffer_get_length(out) > 0) {
 288                         return 0;
 289                 } else {
 290                         bufferevent_free(self);
 291                         return 1;
 292                 }
 293         `}
 294
 295         # The output buffer associated to `self`
 296         fun output_buffer: OutputNativeEvBuffer `{ return bufferevent_get_output(self); `}
 297
 298         # The input buffer associated to `self`
 299         fun input_buffer: InputNativeEvBuffer `{ return bufferevent_get_input(self); `}
 300
 301         # Read data from this buffer
 302         fun read_buffer(buf: NativeEvBuffer): Int `{ return bufferevent_read_buffer(self, buf); `}
 303
 304         # Write data to this buffer
 305         fun write_buffer(buf: NativeEvBuffer): Int `{ return bufferevent_write_buffer(self, buf); `}
 306 end
 307
 308 # A single buffer
 309 extern class NativeEvBuffer `{ struct evbuffer * `}
 310         # Length of data in this buffer
 311         fun length: Int `{ return evbuffer_get_length(self); `}
 312
 313         # Read data from an evbuffer and drain the bytes read
 314         fun remove(buffer: NativeString, len: Int) `{
 315                 evbuffer_remove(self, buffer, len);
 316         `}
 317 end
 318
 319 # An input buffer
 320 extern class InputNativeEvBuffer
 321         super NativeEvBuffer
 322
 323         # Empty/clear `length` data from buffer
 324         fun drain(length: Int) `{ evbuffer_drain(self, length); `}
 325 end
 326
 327 # An output buffer
 328 extern class OutputNativeEvBuffer
 329         super NativeEvBuffer
 330
 331         # Add file to buffer
 332         fun add_file(fd, offset, length: Int): Bool `{
 333                 return evbuffer_add_file(self, fd, offset, length);
 334         `}
 335 end
 336
 337 # A listener acting on an interface and port, spawns `Connection` on new connections
 338 extern class ConnectionListener `{ struct evconnlistener * `}
 339
 340         private new bind_to(base: NativeEventBase, address: NativeString, port: Int, factory: ConnectionFactory)
 341         import ConnectionFactory.accept_connection, error_callback `{
 342
 343                 struct sockaddr_in sin;
 344                 struct evconnlistener *listener;
 345                 ConnectionFactory_incr_ref(factory);
 346
 347                 struct hostent *hostent = gethostbyname(address);
 348
 349                 if (!hostent) {
 350                         return NULL;
 351                 }
 352
 353                 memset(&sin, 0, sizeof(sin));
 354                 sin.sin_family = hostent->h_addrtype;
 355                 sin.sin_port = htons(port);
 356                 memcpy( &(sin.sin_addr.s_addr), (const void*)hostent->h_addr, hostent->h_length );
 357
 358                 listener = evconnlistener_new_bind(base,
 359                         (evconnlistener_cb)accept_connection_cb, factory,
 360                         LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, -1,
 361                         (struct sockaddr*)&sin, sizeof(sin));
 362
 363                 if (listener != NULL) {
 364                         evconnlistener_set_error_cb(listener, (evconnlistener_errorcb)ConnectionListener_error_callback);
 365                 }
 366
 367                 return listener;
 368         `}
 369
 370         # Get the `NativeEventBase` associated to `self`
 371         fun base: NativeEventBase `{ return evconnlistener_get_base(self); `}
 372
 373         # Callback method on listening error
 374         fun error_callback do
 375                 var cstr = socket_error
 376                 sys.stderr.write "libevent error: '{cstr}'"
 377         end
 378
 379         # Error with sockets
 380         fun socket_error: NativeString `{
 381                 // TODO move to Nit and maybe NativeEventBase
 382                 int err = EVUTIL_SOCKET_ERROR();
 383                 return evutil_socket_error_to_string(err);
 384         `}
 385 end
 386
 387 # Factory to listen on sockets and create new `Connection`
 388 class ConnectionFactory
 389         # The `NativeEventBase` for the dispatch loop of this factory
 390         var event_base: NativeEventBase
 391
 392         # Accept a connection on `listener`
 393         #
 394         # By default, it creates a new NativeBufferEvent and calls `spawn_connection`.
 395         fun accept_connection(listener: ConnectionListener, fd: Int, address: Pointer, socklen: Int)
 396         do
 397                 var base = listener.base
 398                 var bev = new NativeBufferEvent.socket(base, fd, bev_opt_close_on_free)
 399                 var conn = spawn_connection(bev)
 400                 bev.enable ev_read|ev_write
 401                 bev.setcb conn
 402         end
 403
 404         # Create a new `Connection` object for `buffer_event`
 405         fun spawn_connection(buffer_event: NativeBufferEvent): Connection
 406         do
 407                 return new Connection(buffer_event)
 408         end
 409
 410         # Listen on `address`:`port` for new connection, which will callback `spawn_connection`
 411         fun bind_to(address: String, port: Int): nullable ConnectionListener
 412         do
 413                 var listener = new ConnectionListener.bind_to(event_base, address.to_cstring, port, self)
 414                 if listener.address_is_null then
 415                         sys.stderr.write "libevent warning: Opening {address}:{port} failed\n"
 416                 end
 417                 return listener
 418         end
 419 end