Class: Iodine::Connection
- Inherits:
-
Object
- Object
- Iodine::Connection
- Defined in:
- lib/iodine/connection.rb,
ext/iodine/iodine_connection.c
Overview
Iodine’s Connection class is the class that TCP/IP, WebSockets and SSE connections inherit from.
Instances of this class are passed to the callback objects. i.e.:
module MyConnectionCallbacks
# called when the callback object is linked with a new client
def on_open client
client.is_a?(Iodine::Connection) # => true
end
# called when data is available
def client, data
client.is_a?(Iodine::Connection) # => true
end
# called when the server is shutting down, before closing the client
# (it's still possible to send messages to the client)
def on_shutdown client
client.is_a?(Iodine::Connection) # => true
end
# called when the client is closed (no longer available)
def on_close client
client.is_a?(Iodine::Connection) # => true
end
# called when all the previous calls to `client.write` have completed
# (the local buffer was drained and is now empty)
def on_drained client
client.is_a?(Iodine::Connection) # => true
end
end
All connection related actions can be performed using the methods provided through this class.
Instance Method Summary collapse
-
#close ⇒ Object
Schedules the connection to be closed.
-
#env ⇒ Object
Returns the connection’s ‘env` (if it originated from an HTTP request).
-
#open? ⇒ Boolean
Returns true if the connection appears to be open (no known issues).
-
#pending ⇒ Object
Returns the number of pending ‘write` operations that need to complete before the next `on_drained` callback is called.
-
#protocol ⇒ Object
Returns the connection’s protocol Symbol (‘:sse`, `:websocket`, etc’).
-
#publish(*args) ⇒ Object
Publishes a message to a channel.
-
#subscribe(*args) ⇒ Object
Subscribes to a Pub/Sub stream / channel or replaces an existing subscription.
-
#timeout ⇒ Object
Returns the timeout / ‘ping` interval for the connection.
-
#timeout=(timeout) ⇒ Object
Sets the timeout / ‘ping` interval for the connection (up to 255 seconds).
-
#unsubscribe(name) ⇒ Object
Unsubscribes from a Pub/Sub stream / channel.
-
#write(data) ⇒ Object
Writes data to the connection asynchronously.
Instance Method Details
#close ⇒ Object
Schedules the connection to be closed.
The connection will be closed once all the scheduled ‘write` operations have been completed (or failed).
230 231 232 233 234 235 236 237 238 239 240 241 |
# File 'ext/iodine/iodine_connection.c', line 230
static VALUE iodine_connection_close(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && !sock_isclosed(c->info.uuid)) {
if (c->info.type == IODINE_CONNECTION_WEBSOCKET) {
websocket_close(c->info.arg);
} else {
sock_close(c->info.uuid);
}
}
return Qnil;
}
|
#env ⇒ Object
Returns the connection’s ‘env` (if it originated from an HTTP request).
318 319 320 321 322 323 324 |
# File 'ext/iodine/iodine_connection.c', line 318
static VALUE iodine_connection_env(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && c->info.env) {
return c->info.env;
}
return Qnil;
}
|
#open? ⇒ Boolean
Returns true if the connection appears to be open (no known issues).
243 244 245 246 247 248 249 |
# File 'ext/iodine/iodine_connection.c', line 243
static VALUE iodine_connection_is_open(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && !sock_isclosed(c->info.uuid)) {
return Qtrue;
}
return Qfalse;
}
|
#pending ⇒ Object
Returns the number of pending ‘write` operations that need to complete before the next `on_drained` callback is called.
254 255 256 257 258 259 260 |
# File 'ext/iodine/iodine_connection.c', line 254
static VALUE iodine_connection_pending(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (!c || sock_isclosed(c->info.uuid)) {
return INT2NUM(-1);
}
return SIZET2NUM((sock_pending(c->info.uuid)));
}
|
#protocol ⇒ Object
Returns the connection’s protocol Symbol (‘:sse`, `:websocket`, etc’).
263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 |
# File 'ext/iodine/iodine_connection.c', line 263
static VALUE iodine_connection_protocol_name(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c) {
switch (c->info.type) {
case IODINE_CONNECTION_WEBSOCKET:
return WebSocketSymbol;
break;
case IODINE_CONNECTION_SSE:
return SSESymbol;
break;
case IODINE_CONNECTION_RAW: /* fallthrough */
return RAWSymbol;
break;
}
}
return Qnil;
}
|
#publish(*args) ⇒ Object
Publishes a message to a channel.
Can be used using two Strings:
publish(to, )
The method accepts an optional ‘engine` argument:
publish(to, , my_pubsub_engine)
Alternatively, accepts the following named arguments:
- :to
-
The channel to publish to (required).
- :message
-
The message to be published (required).
- :engine
-
If provided, the engine to use for pub/sub. Otherwise the default
engine is used.
608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 |
# File 'ext/iodine/iodine_connection.c', line 608
static VALUE iodine_pubsub_publish(int argc, VALUE *argv, VALUE self) {
VALUE rb_ch, rb_msg, rb_engine = Qnil;
const pubsub_engine_s *engine = NULL;
switch (argc) {
case 3:
/* fallthrough */
rb_engine = argv[2];
case 2:
rb_ch = argv[0];
rb_msg = argv[1];
break;
case 1: {
/* single argument must be a Hash */
Check_Type(argv[0], T_HASH);
rb_ch = rb_hash_aref(argv[0], to_id);
if (rb_ch == Qnil || rb_ch == Qfalse) {
rb_ch = rb_hash_aref(argv[0], channel_id);
}
rb_msg = rb_hash_aref(argv[0], message_id);
rb_engine = rb_hash_aref(argv[0], engine_id);
} break;
default:
rb_raise(rb_eArgError, "method accepts 1-3 arguments.");
}
if (rb_msg == Qnil || rb_msg == Qfalse) {
rb_raise(rb_eArgError, "message is required.");
}
Check_Type(rb_msg, T_STRING);
if (rb_ch == Qnil || rb_ch == Qfalse)
rb_raise(rb_eArgError, "target / channel is required .");
if (TYPE(rb_ch) == T_SYMBOL)
rb_ch = rb_sym2str(rb_ch);
Check_Type(rb_ch, T_STRING);
if (rb_engine == Qfalse) {
engine = PUBSUB_PROCESS_ENGINE;
} else if (rb_engine != Qnil) {
// collect engine object
iodine_pubsub_s *e = iodine_pubsub_CData(rb_engine);
if (e) {
engine = e->engine;
}
}
FIOBJ ch = fiobj_str_new(RSTRING_PTR(rb_ch), RSTRING_LEN(rb_ch));
FIOBJ msg = fiobj_str_new(RSTRING_PTR(rb_msg), RSTRING_LEN(rb_msg));
intptr_t ret =
pubsub_publish(.engine = engine, .channel = ch, .message = msg);
fiobj_free(ch);
fiobj_free(msg);
if (!ret)
return Qfalse;
return Qtrue;
(void)self;
}
|
#subscribe(*args) ⇒ Object
Subscribes to a Pub/Sub stream / channel or replaces an existing subscription.
The method accepts 1-2 arguments and an optional block. These are all valid ways to call the method:
subscribe("my_stream") {|source, msg| p msg }
subscribe("my_strea*", match: :redis) {|source, msg| p msg }
subscribe(to: "my_stream") {|source, msg| p msg }
# or use any object that answers `#call(source, msg)`
MyProc = Proc.new {|source, msg| p msg }
subscribe to: "my_stream", match: :redis, handler: MyProc
The first argument must be either a String or a Hash.
The second, optional, argument must be a Hash (if given).
The options Hash supports the following possible keys (other keys are ignored, all keys are Symbols):
- :match
-
The channel / subject name matching type to be used. Valid value is: ‘:redis`. Future versions hope to support `:nats` and `:rabbit` patern matching as well.
- :to
-
The channel / subject to subscribe to.
- :as
-
(only for WebSocket connections) accepts the optional value ‘:binary`. default is `:text`. Note that binary transmissions are illegal for some connections (such as SSE) and an attempted binary subscription will fail for these connections.
- :handler
-
Any object that answers ‘#call(source, msg)` where source is the stream / channel name.
Note: if an existing subscription with the same name exists, it will be replaced by this new subscription.
Returns the name of the subscription, which matches the name be used in #unsubscribe (or nil on failure).
499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 |
# File 'ext/iodine/iodine_connection.c', line 499
static VALUE iodine_pubsub_subscribe(int argc, VALUE *argv, VALUE self) {
// clang-format on
iodine_sub_args_s args = iodine_subscribe_args(argc, argv);
if (args.channel == Qnil) {
return Qnil;
}
iodine_connection_data_s *c = NULL;
if (TYPE(self) == T_MODULE) {
if (!args.block) {
rb_raise(rb_eArgError,
"block or :handler required for local subscriptions.");
}
} else {
c = iodine_connection_validate_data(self);
if (!c || (c->info.type == IODINE_CONNECTION_SSE && args.binary)) {
if (args.block) {
IodineStore.remove(args.block);
}
return Qnil; /* cannot subscribe a closed connection. */
}
if (args.block == Qnil && args.binary) {
args.block = Qtrue;
}
spn_add(&c->ref, 1);
}
FIOBJ channel =
fiobj_str_new(RSTRING_PTR(args.channel), RSTRING_LEN(args.channel));
pubsub_sub_pt sub =
pubsub_subscribe(.channel = channel, .on_message = iodine_on_pubsub,
.on_unsubscribe = iodine_on_unsubscribe, .udata1 = c,
.udata2 = (void *)args.block,
.use_pattern = args.pattern);
fiobj_free(channel);
if (c) {
spn_lock(&c->lock);
if (c->info.uuid == -1) {
pubsub_unsubscribe(sub);
spn_unlock(&c->lock);
return Qnil;
} else {
iodine_sub_add(&c->subscriptions, sub);
}
spn_unlock(&c->lock);
} else {
spn_lock(&sub_lock);
iodine_sub_add(&sub_global, sub);
spn_unlock(&sub_lock);
}
return args.channel;
}
|
#timeout ⇒ Object
Returns the timeout / ‘ping` interval for the connection.
Returns nil on error.
286 287 288 289 290 291 292 293 |
# File 'ext/iodine/iodine_connection.c', line 286
static VALUE iodine_connection_timeout_get(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && !sock_isclosed(c->info.uuid)) {
size_t tout = (size_t)facil_get_timeout(c->info.uuid);
return SIZET2NUM(tout);
}
return Qnil;
}
|
#timeout=(timeout) ⇒ Object
Sets the timeout / ‘ping` interval for the connection (up to 255 seconds).
Returns nil on error.
300 301 302 303 304 305 306 307 308 309 310 311 312 313 |
# File 'ext/iodine/iodine_connection.c', line 300
static VALUE iodine_connection_timeout_set(VALUE self, VALUE timeout) {
Check_Type(timeout, T_FIXNUM);
int tout = NUM2INT(timeout);
if (tout < 0 || tout > 255) {
rb_raise(rb_eRangeError, "timeout out of range.");
return Qnil;
}
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && !sock_isclosed(c->info.uuid)) {
facil_set_timeout(c->info.uuid, (uint8_t)tout);
return timeout;
}
return Qnil;
}
|
#unsubscribe(name) ⇒ Object
Unsubscribes from a Pub/Sub stream / channel.
The method accepts a single arguments, the name used for the subscription. i.e.:
subscribe("my_stream") {|source, msg| p msg }
unsubscribe("my_stream")
Returns ‘true` if the subscription was found.
Returns ‘false` if the subscription didn’t exist.
564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 |
# File 'ext/iodine/iodine_connection.c', line 564
static VALUE iodine_pubsub_unsubscribe(VALUE self, VALUE name) {
// clang-format on
iodine_connection_data_s *c = NULL;
FIOBJ channel = fiobj_str_new(RSTRING_PTR(name), RSTRING_LEN(name));
VALUE ret;
if (TYPE(self) == T_MODULE) {
spn_lock(&sub_lock);
ret = iodine_sub_unsubscribe(&sub_global, channel);
spn_unlock(&sub_lock);
} else {
c = iodine_connection_validate_data(self);
if (!c) {
return Qnil; /* cannot subscribe a closed connection. */
}
spn_lock(&sub_lock);
ret = iodine_sub_unsubscribe(&sub_global, channel);
spn_unlock(&sub_lock);
}
fiobj_free(channel);
return ret;
}
|
#write(data) ⇒ Object
Writes data to the connection asynchronously.
In effect, the ‘write` call does nothing, it only schedules the data to be sent and marks the data as pending.
Use #pending to test how many ‘write` operations are pending completion (`on_drained(client)` will be called when they complete).
177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 |
# File 'ext/iodine/iodine_connection.c', line 177
static VALUE iodine_connection_write(VALUE self, VALUE data) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (!c || sock_isclosed(c->info.uuid)) {
rb_raise(rb_eIOError, "Connection closed or invalid.");
}
switch (c->info.type) {
case IODINE_CONNECTION_WEBSOCKET:
/* WebSockets*/
websocket_write(c->info.arg, RSTRING_PTR(data), RSTRING_LEN(data),
rb_enc_get(data) == IodineUTF8Encoding);
return Qtrue;
break;
case IODINE_CONNECTION_SSE:
/* SSE */
#if 1
http_sse_write(c->info.arg, .data = {.data = RSTRING_PTR(data),
.len = RSTRING_LEN(data)});
return Qtrue;
#else
if (rb_enc_get(data) == IodineUTF8Encoding) {
http_sse_write(c->info.arg, .data = {.data = RSTRING_PTR(data),
.len = RSTRING_LEN(data)});
return Qtrue;
}
fprintf(stderr, "WARNING: ignoring an attept to write binary data to "
"non-binary protocol (SSE).\n");
return Qfalse;
// rb_raise(rb_eEncodingError,
// "This Connection type requires data to be UTF-8 encoded.");
#endif
break;
case IODINE_CONNECTION_RAW: /* fallthrough */
default: {
size_t len = RSTRING_LEN(data);
char *copy = fio_malloc(len);
if (!copy) {
rb_raise(rb_eNoMemError, "failed to allocate memory for network buffer!");
}
memcpy(copy, RSTRING_PTR(data), len);
sock_write2(.uuid = c->info.uuid, .buffer = copy, .length = len,
.dealloc = fio_free);
return Qtrue;
} break;
}
return Qnil;
}
|