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.

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.



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

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`.



179
180
181
182
183
184
185
186
187
188
189
 
# File 'ext/iodine/iodine_core.c', line 179

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`.



217
218
219
220
221
222
223
224
225
226
227
228
 
# File 'ext/iodine/iodine_core.c', line 217

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.



271
 
# File 'ext/iodine/iodine_core.c', line 271

static VALUE not_implemented(VALUE 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.



282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'ext/iodine/iodine_core.c', line 282

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.



273
 
# File 'ext/iodine/iodine_core.c', line 273

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

#on_open ⇒ Object

implement this callback to handle the event.



271
 
# File 'ext/iodine/iodine_core.c', line 271

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

#on_ready ⇒ Object

implement this callback to handle the event.



271
 
# File 'ext/iodine/iodine_core.c', line 271

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

#on_shutdown ⇒ Object

implement this callback to handle the event.



271
 
# File 'ext/iodine/iodine_core.c', line 271

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

#ping ⇒ Object

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



266
267
268
269
 
# File 'ext/iodine/iodine_core.c', line 266

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 te 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;
}

#upgrade(handler) ⇒ Object 



256
257
258
 
# File 'ext/iodine/iodine_core.c', line 256

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;
}