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