Class: Iodine::Connection
- Inherits:
-
Data
- Object
- Data
- 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).
217 218 219 220 221 222 223 224 225 226 227 228 |
# File 'ext/iodine/iodine_connection.c', line 217
static VALUE iodine_connection_close(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && !fio_is_closed(c->info.uuid)) {
if (c->info.type == IODINE_CONNECTION_WEBSOCKET) {
websocket_close(c->info.arg); /* sends WebSocket close packet */
} else {
fio_close(c->info.uuid);
}
}
return Qnil;
}
|
#env ⇒ Object
Returns the connection’s ‘env` (if it originated from an HTTP request).
308 309 310 311 312 313 314 |
# File 'ext/iodine/iodine_connection.c', line 308
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).
230 231 232 233 234 235 236 |
# File 'ext/iodine/iodine_connection.c', line 230
static VALUE iodine_connection_is_open(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && !fio_is_closed(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`).
244 245 246 247 248 249 250 |
# File 'ext/iodine/iodine_connection.c', line 244
static VALUE iodine_connection_pending(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (!c || fio_is_closed(c->info.uuid)) {
return INT2NUM(-1);
}
return SIZET2NUM((fio_pending(c->info.uuid)));
}
|
#protocol ⇒ Object
Returns the connection’s protocol Symbol (‘:sse`, `:websocket`, etc’).
253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 |
# File 'ext/iodine/iodine_connection.c', line 253
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.
588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 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 |
# File 'ext/iodine/iodine_connection.c', line 588
static VALUE iodine_pubsub_publish(int argc, VALUE *argv, VALUE self) {
// clang-format on
VALUE rb_ch, rb_msg, rb_engine = Qnil;
const fio_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 = FIO_PUBSUB_PROCESS;
} else if (rb_engine != Qnil) {
// collect engine object
iodine_pubsub_s *e = iodine_pubsub_CData(rb_engine);
if (e) {
engine = e->engine;
}
}
fio_publish(.engine = engine, .channel = IODINE_RSTRINFO(rb_ch),
.message = IODINE_RSTRINFO(rb_msg));
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).
484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 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 |
# File 'ext/iodine/iodine_connection.c', line 484
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;
}
fio_atomic_add(&c->ref, 1);
}
subscription_s *sub =
fio_subscribe(.channel = IODINE_RSTRINFO(args.channel),
.on_message = iodine_on_pubsub,
.on_unsubscribe = iodine_on_unsubscribe, .udata1 = c,
.udata2 = (void *)args.block, .match = args.pattern);
if (c) {
fio_lock(&c->lock);
if (c->info.uuid == -1) {
fio_unsubscribe(sub);
fio_unlock(&c->lock);
return Qnil;
} else {
iodine_sub_add(&c->subscriptions, sub);
}
fio_unlock(&c->lock);
} else {
fio_lock(&sub_lock);
iodine_sub_add(&sub_global, sub);
fio_unlock(&sub_lock);
}
return args.channel;
}
|
#timeout ⇒ Object
Returns the timeout / ‘ping` interval for the connection.
Returns nil on error.
276 277 278 279 280 281 282 283 |
# File 'ext/iodine/iodine_connection.c', line 276
static VALUE iodine_connection_timeout_get(VALUE self) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (c && !fio_is_closed(c->info.uuid)) {
size_t tout = (size_t)fio_timeout_get(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.
290 291 292 293 294 295 296 297 298 299 300 301 302 303 |
# File 'ext/iodine/iodine_connection.c', line 290
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 && !fio_is_closed(c->info.uuid)) {
fio_timeout_set(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.
546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 |
# File 'ext/iodine/iodine_connection.c', line 546
static VALUE iodine_pubsub_unsubscribe(VALUE self, VALUE name) {
// clang-format on
iodine_connection_data_s *c = NULL;
VALUE ret;
if (TYPE(self) == T_MODULE) {
fio_lock(&sub_lock);
ret = iodine_sub_unsubscribe(&sub_global, IODINE_RSTRINFO(name));
fio_unlock(&sub_lock);
} else {
c = iodine_connection_validate_data(self);
if (!c) {
return Qnil; /* cannot unsubscribe a closed connection. */
}
fio_lock(&sub_lock);
ret = iodine_sub_unsubscribe(&sub_global, IODINE_RSTRINFO(name));
fio_unlock(&sub_lock);
}
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).
170 171 172 173 174 175 176 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 |
# File 'ext/iodine/iodine_connection.c', line 170
static VALUE iodine_connection_write(VALUE self, VALUE data) {
iodine_connection_data_s *c = iodine_connection_validate_data(self);
if (!c || fio_is_closed(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, IODINE_RSTRINFO(data),
rb_enc_get(data) == IodineUTF8Encoding);
return Qtrue;
break;
case IODINE_CONNECTION_SSE:
/* SSE */
#if 1
http_sse_write(c->info.arg, .data = IODINE_RSTRINFO(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: {
fio_write(c->info.uuid, RSTRING_PTR(data), RSTRING_LEN(data));
return Qtrue;
} break;
}
return Qnil;
}
|