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 <event2/listener.h>
  26         #include <event2/bufferevent.h>
  27         #include <event2/buffer.h>
  28 `}
  29
  30 in "C" `{
  31         #include <sys/stat.h>
  32         #include <sys/types.h>
  33         #include <fcntl.h>
  34         #include <errno.h>
  35         #include <string.h>
  36
  37         #include <sys/socket.h>
  38         #include <arpa/inet.h>
  39         #include <netinet/in.h>
  40         #include <netinet/ip.h>
  41
  42 // Protect callbacks for compatibility with light FFI
  43 #ifdef Connection_decr_ref
  44         // Callback forwarded to 'Connection.write_callback'
  45         static void c_write_cb(struct bufferevent *bev, Connection ctx) {
  46                 Connection_write_callback(ctx);
  47         }
  48
  49         // Callback forwarded to 'Connection.read_callback_native'
  50         static void c_read_cb(struct bufferevent *bev, Connection ctx)
  51         {
  52                 Connection_read_callback_native(ctx, bev);
  53         }
  54
  55         // Callback forwarded to 'Connection.event_callback'
  56         static void c_event_cb(struct bufferevent *bev, short events, Connection ctx)
  57         {
  58                 int release = Connection_event_callback(ctx, events);
  59                 if (release) Connection_decr_ref(ctx);
  60         }
  61
  62         // Callback forwarded to 'ConnectionFactory.accept_connection'
  63         static void accept_connection_cb(struct evconnlistener *listener, evutil_socket_t fd,
  64                 struct sockaddr *addrin, int socklen, ConnectionFactory ctx)
  65         {
  66                 ConnectionFactory_accept_connection(ctx, listener, fd, addrin, socklen);
  67         }
  68 #endif
  69
  70 `}
  71
  72 # Structure to hold information and state for a Libevent dispatch loop.
  73 #
  74 # The event_base lies at the center of Libevent; every application will
  75 # have one.  It keeps track of all pending and active events, and
  76 # notifies your application of the active ones.
  77 extern class NativeEventBase `{ struct event_base * `}
  78
  79         # Create a new event_base to use with the rest of Libevent
  80         new `{ return event_base_new(); `}
  81
  82         # Has `self` been correctly initialized?
  83         fun is_valid: Bool do return not address_is_null
  84
  85         # Event dispatching loop
  86         #
  87         # This loop will run the event base until either there are no more added
  88         # events, or until something calls `exit_loop`.
  89         fun dispatch `{ event_base_dispatch(self); `}
  90
  91         # Exit the event loop
  92         #
  93         # TODO support timer
  94         fun exit_loop `{ event_base_loopexit(self, NULL); `}
  95
  96         # Destroy this instance
  97         fun destroy `{ event_base_free(self); `}
  98 end
  99
 100 # Spawned to manage a specific connection
 101 #
 102 # TODO, use polls
 103 class Connection
 104         super Writer
 105
 106         # Closing this connection has been requested, but may not yet be `closed`
 107         var close_requested = false
 108
 109         # This connection is closed
 110         var closed = false
 111
 112         # The native libevent linked to `self`
 113         var native_buffer_event: NativeBufferEvent
 114
 115         # Close this connection if possible, otherwise mark it to be closed later
 116         redef fun close
 117         do
 118                 if closed then return
 119
 120                 var i = native_buffer_event.input_buffer
 121                 var o = native_buffer_event.output_buffer
 122                 if i.length > 0 or o.length > 0 then
 123                         close_requested = true
 124                 else
 125                         force_close
 126                 end
 127         end
 128
 129         # Force closing this connection and freeing `native_buffer_event`
 130         fun force_close
 131         do
 132                 if closed then return
 133
 134                 native_buffer_event.free
 135                 closed = true
 136         end
 137
 138         # Callback method on a write event
 139         fun write_callback
 140         do
 141                 if close_requested then close
 142         end
 143
 144         private fun read_callback_native(bev: NativeBufferEvent)
 145         do
 146                 var evbuffer = bev.input_buffer
 147                 var len = evbuffer.length
 148                 var buf = new NativeString(len)
 149                 evbuffer.remove(buf, len)
 150                 var str = buf.to_s_with_length(len)
 151                 read_callback str
 152         end
 153
 154         # Callback method when data is available to read
 155         fun read_callback(content: String)
 156         do
 157                 if close_requested then close
 158         end
 159
 160         # Callback method on events: EOF, user-defined timeout and unrecoverable errors
 161         #
 162         # Returns `true` if the native handles to `self` can be released.
 163         fun event_callback(events: Int): Bool
 164         do
 165                 if events & bev_event_error != 0 or events & bev_event_eof != 0 then
 166                         if events & bev_event_error != 0 then
 167                                 var sock_err = evutil_socket_error
 168                                 # Ignore some normal errors and print the others for debugging
 169                                 if sock_err == 110 then
 170                                         # Connection timed out (ETIMEDOUT)
 171                                 else if sock_err == 104 then
 172                                         # Connection reset by peer (ECONNRESET)
 173                                 else
 174                                         print_error "libevent error event: {evutil_socket_error_to_string(sock_err)} ({sock_err})"
 175                                 end
 176                         end
 177                         force_close
 178                         return true
 179                 end
 180
 181                 return false
 182         end
 183
 184         # Write a string to the connection
 185         redef fun write(str)
 186         do
 187                 if close_requested then return
 188                 native_buffer_event.write(str.to_cstring, str.byte_length)
 189         end
 190
 191         redef fun write_byte(byte)
 192         do
 193                 if close_requested then return
 194                 native_buffer_event.write_byte(byte)
 195         end
 196
 197         redef fun write_bytes(bytes)
 198         do
 199                 if close_requested then return
 200                 native_buffer_event.write(bytes.items, bytes.length)
 201         end
 202
 203         # Write a file to the connection
 204         #
 205         # If `not path.file_exists`, the method returns.
 206         fun write_file(path: String)
 207         do
 208                 if close_requested then return
 209
 210                 var file = new FileReader.open(path)
 211                 if file.last_error != null then
 212                         var error = new IOError("Failed to open file at '{path}'")
 213                         error.cause = file.last_error
 214                         self.last_error = error
 215                         file.close
 216                         return
 217                 end
 218
 219                 var stat = file.file_stat
 220                 if stat == null then
 221                         last_error = new IOError("Failed to stat file at '{path}'")
 222                         file.close
 223                         return
 224                 end
 225
 226                 var err = native_buffer_event.output_buffer.add_file(file.fd, 0, stat.size)
 227                 if err then
 228                         last_error = new IOError("Failed to add file at '{path}'")
 229                         file.close
 230                 end
 231         end
 232 end
 233
 234 # ---
 235 # Error code for event callbacks
 236
 237 # error encountered while reading
 238 fun bev_event_reading: Int `{ return BEV_EVENT_READING; `}
 239
 240 # error encountered while writing
 241 fun bev_event_writing: Int `{ return BEV_EVENT_WRITING; `}
 242
 243 # eof file reached
 244 fun bev_event_eof: Int `{ return BEV_EVENT_EOF; `}
 245
 246 # unrecoverable error encountered
 247 fun bev_event_error: Int `{ return BEV_EVENT_ERROR; `}
 248
 249 # user-specified timeout reached
 250 fun bev_event_timeout: Int `{ return BEV_EVENT_TIMEOUT; `}
 251
 252 # connect operation finished.
 253 fun bev_event_connected: Int `{ return BEV_EVENT_CONNECTED; `}
 254
 255 # Global error code for the last socket operation on the calling thread
 256 #
 257 # Not idempotent on all platforms.
 258 fun evutil_socket_error: Int `{
 259         return EVUTIL_SOCKET_ERROR();
 260 `}
 261
 262 # Convert an error code from `evutil_socket_error` to a string
 263 fun evutil_socket_error_to_string(error_code: Int): NativeString `{
 264         return evutil_socket_error_to_string(error_code);
 265 `}
 266
 267 # ---
 268 # Options that can be specified when creating a `NativeBufferEvent`
 269
 270 # Close the underlying file descriptor/bufferevent/whatever when this bufferevent is freed.
 271 fun bev_opt_close_on_free: Int `{ return BEV_OPT_CLOSE_ON_FREE; `}
 272
 273 # If threading is enabled, protect the operations on this bufferevent with a lock.
 274 fun bev_opt_threadsafe: Int `{ return BEV_OPT_THREADSAFE; `}
 275
 276 # Run callbacks deferred in the event loop.
 277 fun bev_opt_defer_callbacks: Int `{ return BEV_OPT_DEFER_CALLBACKS; `}
 278
 279 # If set, callbacks are executed without locks being held on the bufferevent.
 280 fun bev_opt_unlock_callbacks: Int `{ return BEV_OPT_UNLOCK_CALLBACKS; `}
 281
 282 # ---
 283 # Options for `NativeBufferEvent::enable`
 284
 285 # Read operation
 286 fun ev_read: Int `{ return EV_READ; `}
 287
 288 # Write operation
 289 fun ev_write: Int `{ return EV_WRITE; `}
 290
 291 # ---
 292
 293 # A buffer event structure, strongly associated to a connection, an input buffer and an output_buffer
 294 extern class NativeBufferEvent `{ struct bufferevent * `}
 295
 296         # Socket-based `NativeBufferEvent` that reads and writes data onto a network
 297         new socket(base: NativeEventBase, fd, options: Int) `{
 298                 return bufferevent_socket_new(base, fd, options);
 299         `}
 300
 301         # Enable a bufferevent.
 302         fun enable(operation: Int) `{
 303                 bufferevent_enable(self, operation);
 304         `}
 305
 306         # Set callbacks to `read_callback_native`, `write_callback` and `event_callback` of `conn`
 307         fun setcb(conn: Connection) import Connection.read_callback_native,
 308         Connection.write_callback, Connection.event_callback, NativeString `{
 309                 Connection_incr_ref(conn);
 310                 bufferevent_setcb(self,
 311                         (bufferevent_data_cb)c_read_cb,
 312                         (bufferevent_data_cb)c_write_cb,
 313                         (bufferevent_event_cb)c_event_cb, conn);
 314         `}
 315
 316         # Write `length` bytes of `line`
 317         fun write(line: NativeString, length: Int): Int `{
 318                 return bufferevent_write(self, line, length);
 319         `}
 320
 321         # Write the byte `value`
 322         fun write_byte(value: Byte): Int `{
 323                 unsigned char byt = (unsigned char)value;
 324                 return bufferevent_write(self, &byt, 1);
 325         `}
 326
 327         redef fun free `{ bufferevent_free(self); `}
 328
 329         # The output buffer associated to `self`
 330         fun output_buffer: OutputNativeEvBuffer `{ return bufferevent_get_output(self); `}
 331
 332         # The input buffer associated to `self`
 333         fun input_buffer: InputNativeEvBuffer `{ return bufferevent_get_input(self); `}
 334
 335         # Read data from this buffer
 336         fun read_buffer(buf: NativeEvBuffer): Int `{ return bufferevent_read_buffer(self, buf); `}
 337
 338         # Write data to this buffer
 339         fun write_buffer(buf: NativeEvBuffer): Int `{ return bufferevent_write_buffer(self, buf); `}
 340 end
 341
 342 # A single buffer
 343 extern class NativeEvBuffer `{ struct evbuffer * `}
 344         # Length of data in this buffer
 345         fun length: Int `{ return evbuffer_get_length(self); `}
 346
 347         # Read data from an evbuffer and drain the bytes read
 348         fun remove(buffer: NativeString, len: Int) `{
 349                 evbuffer_remove(self, buffer, len);
 350         `}
 351 end
 352
 353 # An input buffer
 354 extern class InputNativeEvBuffer
 355         super NativeEvBuffer
 356
 357         # Empty/clear `length` data from buffer
 358         fun drain(length: Int) `{ evbuffer_drain(self, length); `}
 359 end
 360
 361 # An output buffer
 362 extern class OutputNativeEvBuffer
 363         super NativeEvBuffer
 364
 365         # Add file to buffer
 366         fun add_file(fd, offset, length: Int): Bool `{
 367                 return evbuffer_add_file(self, fd, offset, length);
 368         `}
 369 end
 370
 371 # A listener acting on an interface and port, spawns `Connection` on new connections
 372 extern class ConnectionListener `{ struct evconnlistener * `}
 373
 374         private new bind_to(base: NativeEventBase, address: NativeString, port: Int, factory: ConnectionFactory)
 375         import ConnectionFactory.accept_connection, error_callback `{
 376
 377                 struct sockaddr_in sin;
 378                 struct evconnlistener *listener;
 379                 ConnectionFactory_incr_ref(factory);
 380
 381                 struct hostent *hostent = gethostbyname(address);
 382
 383                 if (!hostent) {
 384                         return NULL;
 385                 }
 386
 387                 memset(&sin, 0, sizeof(sin));
 388                 sin.sin_family = hostent->h_addrtype;
 389                 sin.sin_port = htons(port);
 390                 memcpy( &(sin.sin_addr.s_addr), (const void*)hostent->h_addr, hostent->h_length );
 391
 392                 listener = evconnlistener_new_bind(base,
 393                         (evconnlistener_cb)accept_connection_cb, factory,
 394                         LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, -1,
 395                         (struct sockaddr*)&sin, sizeof(sin));
 396
 397                 if (listener != NULL) {
 398                         evconnlistener_set_error_cb(listener, (evconnlistener_errorcb)ConnectionListener_error_callback);
 399                 }
 400
 401                 return listener;
 402         `}
 403
 404         # Get the `NativeEventBase` associated to `self`
 405         fun base: NativeEventBase `{ return evconnlistener_get_base(self); `}
 406
 407         # Callback method on listening error
 408         fun error_callback do
 409                 var cstr = evutil_socket_error_to_string(evutil_socket_error)
 410                 print_error "libevent error: '{cstr}'"
 411         end
 412 end
 413
 414 # Factory to listen on sockets and create new `Connection`
 415 class ConnectionFactory
 416         # The `NativeEventBase` for the dispatch loop of this factory
 417         var event_base: NativeEventBase
 418
 419         # Accept a connection on `listener`
 420         #
 421         # By default, it creates a new NativeBufferEvent and calls `spawn_connection`.
 422         fun accept_connection(listener: ConnectionListener, fd: Int, addrin: Pointer, socklen: Int)
 423         do
 424                 var base = listener.base
 425                 var bev = new NativeBufferEvent.socket(base, fd, bev_opt_close_on_free)
 426
 427                 # Human representation of remote client address
 428                 var addr_len = 46 # Longest possible IPv6 address + null byte
 429                 var addr_buf = new NativeString(addr_len)
 430                 addr_buf = addrin_to_address(addrin, addr_buf, addr_len)
 431                 var addr = if addr_buf.address_is_null then
 432                                 "Unknown address"
 433                         else addr_buf.to_s
 434
 435                 var conn = spawn_connection(bev, addr)
 436                 bev.enable ev_read|ev_write
 437                 bev.setcb conn
 438         end
 439
 440         # Create a new `Connection` object for `buffer_event`
 441         fun spawn_connection(buffer_event: NativeBufferEvent, address: String): Connection
 442         do
 443                 return new Connection(buffer_event)
 444         end
 445
 446         # Listen on `address`:`port` for new connection, which will callback `spawn_connection`
 447         fun bind_to(address: String, port: Int): nullable ConnectionListener
 448         do
 449                 var listener = new ConnectionListener.bind_to(event_base, address.to_cstring, port, self)
 450                 if listener.address_is_null then
 451                         sys.stderr.write "libevent warning: Opening {address}:{port} failed\n"
 452                 end
 453                 return listener
 454         end
 455
 456         # Put string representation of source `address` into `buf`
 457         private fun addrin_to_address(address: Pointer, buf: NativeString, buf_len: Int): NativeString `{
 458                 struct sockaddr *addrin = (struct sockaddr*)address;
 459
 460                 if (addrin->sa_family == AF_INET) {
 461                         struct in_addr *src = &((struct sockaddr_in*)addrin)->sin_addr;
 462                         return (char *)inet_ntop(addrin->sa_family, src, buf, buf_len);
 463                 }
 464                 else if (addrin->sa_family == AF_INET6) {
 465                         struct in6_addr *src = &((struct sockaddr_in6*)addrin)->sin6_addr;
 466                         return (char *)inet_ntop(addrin->sa_family, src, buf, buf_len);
 467                 }
 468                 return NULL;
 469         `}
 470 end