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
# Allows the module to be used as a static callback object (avoiding object allocation)
extend self
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).
235 236 237 238 239 240 241 242 243 244 245 246 |
# File 'ext/iodine/iodine_connection.c', line 235
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); /* sends WebSocket close packet */
} else {
sock_close(c->info.uuid);
}
}
return Qnil;
}
|
#env ⇒ Object
Returns the connection’s ‘env` (if it originated from an HTTP request).
326 327 328 329 330 331 332 |
# File 'ext/iodine/iodine_connection.c', line 326
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).
248 249 250 251 252 253 254 |
# File 'ext/iodine/iodine_connection.c', line 248
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.
Returns -1 if the connection is closed and 0 if ‘on_drained` won’t be scheduled (no pending ‘write`).
262 263 264 265 266 267 268 |
# File 'ext/iodine/iodine_connection.c', line 262
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’).
271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 |
# File 'ext/iodine/iodine_connection.c', line 271
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.
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 666 667 668 669 670 671 672 673 674 |
# File 'ext/iodine/iodine_connection.c', line 616
static VALUE iodine_pubsub_publish(int argc, VALUE *argv, VALUE self) {
// clang-format on
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).
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 550 551 552 553 554 555 556 557 |
# File 'ext/iodine/iodine_connection.c', line 507
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.
294 295 296 297 298 299 300 301 |
# File 'ext/iodine/iodine_connection.c', line 294
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.
308 309 310 311 312 313 314 315 316 317 318 319 320 321 |
# File 'ext/iodine/iodine_connection.c', line 308
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.
572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 |
# File 'ext/iodine/iodine_connection.c', line 572
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).
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 223 224 225 226 227 |
# File 'ext/iodine/iodine_connection.c', line 180
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)) {
// don't throw exceptions - closed connections are unavoidable.
return Qnil;
// 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;
}
|