Module: Iodine::Protocol

Defined in:
lib/iodine/protocol.rb,
ext/iodine/iodine_core.c

Overview

The Protocol class is used only for documenting the Protocol API, it will not be included when requiring ‘iodine`.

A dynamic (stateful) protocol is defined as a Ruby class instance which is in control of one single connection.

It is called dynamic because it is dynamically allocated for each connection and then discarded, also it sounded better then calling it “the stateful protocol”, even though that’s what it actually is.

It is (mostly) thread-safe as long as it’s operations are limited to the scope of the object.

The Callbacks

A protocol class MUST contain ONE of the following callbacks:

on_data

called whened there’s data available to be read, but no data was read just yet. ‘on_data` will not be called again untill all the existing network buffer was read (edge triggered event).

on_message(buffer)

the default ‘on_data` implementation creates a 1Kb buffer and reads data while recycling the same String memory space. The buffer is forwarded to the `on_message` callback before being recycled. The buffer object will be over-written once `on_message` returns, so creating a persistent copy requires `buffer.dup`.

A protocol class MAY contain any of the following optional callbacks:

on_open

called after a new connection was accepted and the protocol was linked with Iodine’s Protocol API. Initialization should be performed here.

ping

called whenever timeout was reached. The default implementation will close the connection unless a protocol task (#defer, ‘on_data` or `on_message`) are busy in the background.

on_shutdown

called if the connection is still open while the server is shutting down. This allows the protocol to send a “going away” frame before the connection is closed and ‘on_close` is called.

on_close

called after a connection was closed, for any cleanup (if any).

WARNING: for thread safety and connection management, ‘on_open`, `on_shutdown`, `on_close` and `ping` will all be performed within the reactor’s main thread. Do not run long running tasks within these callbacks, or the server might block while you do. Use #defer to run protocol related tasks (this locks the connection, preventing it from running more then one task at a time and offering thread safety), or #run to run asynchronous tasks that aren’t protocol related.

Connection timeouts

By setting a class variable called ‘@timeout` it is possible to define a default timeout for new connections. However, changing this default timeout should be performed using the #timeout methods.

The API:

After a new connection is accepted and a new protocol object is created, the protocol will be linked with Iodine’s Protocol API. Only the main protocol will be able to access the API within ‘initialize`, so it’s best to use ‘on_open` for any Initialization required.

Instance Method Summary collapse

Instance Method Details

#close ⇒ Object

Closes a connection.

The connection will be closed only once all the data was sent.

Returns self.



173
174
175
176
177
 
# File 'ext/iodine/iodine_core.c', line 173

static VALUE dyn_close(VALUE self) {
  intptr_t fd = iodine_get_fd(self);
  sock_close(fd);
  return self;
}

#defer ⇒ Object

Runs the required block later (defers the blocks execution).

Unlike Iodine#run, the block will *not run concurrently with any other callback for this object (except ‘ping` and `on_ready`).

Also, unlike Iodine#run, the block will not be called unless the connection remains open at the time it’s execution is scheduled.

Always returns ‘self`.



206
207
208
209
210
211
212
213
214
215
216
 
# File 'ext/iodine/iodine_core.c', line 206

static VALUE dyn_defer(VALUE self) {
  // requires a block to be passed
  rb_need_block();
  VALUE block = rb_block_proc();
  if (block == Qnil)
    return Qfalse;
  Registry.add(block);
  intptr_t fd = iodine_get_fd(self);
  server_task(fd, dyn_perform_defer, (void *)block, dyn_defer_fallback);
  return self;
}

#each ⇒ Object

Runs the required block for each dynamic protocol connection except this one.

Tasks will be performed within each connections lock, so no connection will have more then one task being performed at the same time (similar to #defer).

Also, unlike Iodine.run, the block will not be called unless the connection remains open at the time it’s execution is scheduled.

Always returns ‘self`.



247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'ext/iodine/iodine_core.c', line 247

static VALUE dyn_each(VALUE self) {
  // requires a block to be passed
  rb_need_block();
  VALUE block = rb_block_proc();
  if (block == Qnil)
    return Qfalse;
  Registry.add(block);
  intptr_t fd = iodine_get_fd(self);
  server_each(fd, iodine_protocol_service, dyn_perform_each_task, (void *)block,
              dyn_finish_each_task);
  return self;
}

#on_close ⇒ Object

implement this callback to handle the event.



301
302
303
304
 
# File 'ext/iodine/iodine_core.c', line 301

static VALUE not_implemented(VALUE self) {
  (void)(self);
  return Qnil;
}

#on_data ⇒ Object

A default on_data implementation will read up to 1Kb into a reusable buffer from the socket and call the ‘on_message` callback.

It is recommended that you implement this callback if messages might require more then 1Kb of space.



319
320
321
322
323
324
325
326
327
328
329
330
331
 
# File 'ext/iodine/iodine_core.c', line 319

static VALUE default_on_data(VALUE self) {
  VALUE buff = rb_ivar_get(self, buff_var_id);
  if (buff == Qnil) {
    rb_ivar_set(self, buff_var_id, (buff = rb_str_buf_new(1024)));
  }
  do {
    dyn_read(1, &buff, self);
    if (!RSTRING_LEN(buff))
      return Qnil;
    rb_funcall(self, on_message_func_id, 1, buff);
  } while (RSTRING_LEN(buff) == rb_str_capacity(buff));
  return Qnil;
}

#on_message(data) ⇒ Object

implement this callback to handle the event.



306
307
308
309
310
 
# File 'ext/iodine/iodine_core.c', line 306

static VALUE not_implemented2(VALUE self, VALUE data) {
  (void)(self);
  (void)(data);
  return Qnil;
}

#on_open ⇒ Object

implement this callback to handle the event.



301
302
303
304
 
# File 'ext/iodine/iodine_core.c', line 301

static VALUE not_implemented(VALUE self) {
  (void)(self);
  return Qnil;
}

#on_ready ⇒ Object

implement this callback to handle the event.



301
302
303
304
 
# File 'ext/iodine/iodine_core.c', line 301

static VALUE not_implemented(VALUE self) {
  (void)(self);
  return Qnil;
}

#on_shutdown ⇒ Object

implement this callback to handle the event.



301
302
303
304
 
# File 'ext/iodine/iodine_core.c', line 301

static VALUE not_implemented(VALUE self) {
  (void)(self);
  return Qnil;
}

#ping ⇒ Object

Implement this callback to handle the event. The default implementation will close the connection.



296
297
298
299
 
# File 'ext/iodine/iodine_core.c', line 296

static VALUE not_implemented_ping(VALUE self) {
  sock_close(iodine_get_fd(self));
  return Qnil;
}

#read(*args) ⇒ Object

Reads ‘n` bytes from the network connection. The number of bytes to be read (n) is:

  • the number of bytes set in the optional ‘buffer_or_length` argument.

  • the String capacity (not length) of the String passed as the optional ‘buffer_or_length` argument.

  • 1024 Bytes (1Kb) if the optional ‘buffer_or_length` is either missing or contains a String who’s capacity is less then 1Kb.

Returns a String (either the same one used as the buffer or a new one) on a successful read. Returns ‘nil` if no data was available.



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'ext/iodine/iodine_core.c', line 63

static VALUE dyn_read(int argc, VALUE *argv, VALUE self) {
  if (argc > 1) {
    rb_raise(
        rb_eArgError,
        "read accepts only one argument - a Fixnum (buffer length) or a String "
        "(it's capacity - or 1Kb, whichever's the higher - will be used as "
        "buffer's length).");
    return Qnil;
  }
  VALUE buffer = (argc == 1 ? argv[0] : Qnil);
  if (buffer != Qnil && TYPE(buffer) != T_FIXNUM && TYPE(buffer) != T_STRING) {
    rb_raise(rb_eTypeError,
             "buffer should either be a length (a new string will be created) "
             "or a string (reading will be limited to the original string's "
             "capacity or 1Kb - whichever the larger).");
    return Qnil;
  }
  VALUE str;
  long len;
  intptr_t fd = iodine_get_fd(self);
  if (buffer == Qnil) {
    buffer = LONG2FIX(1024);
  }
  if (TYPE(buffer) == T_FIXNUM) {
    len = FIX2LONG(buffer);
    if (len <= 0)
      len = 1024;
    str = rb_str_buf_new(len);
    // create a rb_String with X length and take it's pointer
    // rb_str_resize(VALUE str, long len)
    // RSTRING_PTR(str)
  } else {
    // take the string's pointer and length
    len = rb_str_capacity(buffer);
    // make sure the string is modifiable
    rb_str_modify(buffer);
    // resize the string if needed.
    if (len < 1024)
      rb_str_resize(buffer, (len = 1024));
    str = buffer;
  }
  ssize_t in = sock_read(fd, RSTRING_PTR(str), len);
  // make sure it's binary encoded
  rb_enc_associate_index(str, BinaryEncodingIndex);
  // set actual size....
  if (in > 0)
    rb_str_set_len(str, (long)in);
  else {
    rb_str_set_len(str, 0);
    str = Qnil;
  }
  // return empty string? or fix above if to return Qnil?
  return str;
}

#timeout ⇒ Object

Returns the connection’s timeout.



159
160
161
162
163
164
 
# File 'ext/iodine/iodine_core.c', line 159

static VALUE dyn_get_timeout(VALUE self) {
  intptr_t fd = iodine_get_fd(self);
  uint8_t tout = server_get_timeout(fd);
  unsigned int tout_int = tout;
  return UINT2NUM(tout_int);
}

#timeout=(timeout) ⇒ Object

Update’s a connection’s timeout.

Returns self.



147
148
149
150
151
152
153
154
 
# File 'ext/iodine/iodine_core.c', line 147

static VALUE dyn_set_timeout(VALUE self, VALUE timeout) {
  intptr_t fd = iodine_get_fd(self);
  unsigned int tout = FIX2UINT(timeout);
  if (tout > 255)
    tout = 255;
  server_set_timeout(fd, tout);
  return self;
}

#upgrade(handler) ⇒ Object 



286
287
288
 
# File 'ext/iodine/iodine_core.c', line 286

VALUE dyn_upgrade(VALUE self, VALUE handler) {
  return iodine_upgrade2basic(iodine_get_fd(self), handler);
}

#write(data) ⇒ Object

Writes data to the connection. Returns ‘false` on error and `self` on success.



121
122
123
124
125
126
 
# File 'ext/iodine/iodine_core.c', line 121

static VALUE dyn_write(VALUE self, VALUE data) {
  intptr_t fd = iodine_get_fd(self);
  if (sock_write(fd, RSTRING_PTR(data), RSTRING_LEN(data)))
    return Qfalse;
  return self;
}

#write_urgent(data) ⇒ Object

Writes data to the connection. The data will be sent as soon as possible without fragmantation of previously scheduled data.

Returns ‘false` on error and `self` on success.



134
135
136
137
138
139
140
 
# File 'ext/iodine/iodine_core.c', line 134

static VALUE dyn_write_urgent(VALUE self, VALUE data) {
  intptr_t fd = iodine_get_fd(self);
  if (sock_write2(.fduuid = fd, .buffer = RSTRING(data),
                  .length = RSTRING_LEN(data), .urgent = 1))
    return Qfalse;
  return self;
}