Module: EventMachine
- Defined in:
- lib/em/future.rb,
lib/em/pool.rb,
lib/em/queue.rb,
lib/em/timers.rb,
lib/em/channel.rb,
lib/em/version.rb,
lib/em/callback.rb,
lib/em/iterator.rb,
lib/em/resolver.rb,
lib/em/streamer.rb,
lib/em/processes.rb,
lib/em/protocols.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/pure_ruby.rb,
lib/em/spawnable.rb,
lib/em/tick_loop.rb,
lib/eventmachine.rb,
lib/em/completion.rb,
lib/em/connection.rb,
lib/em/deferrable.rb,
lib/em/file_watch.rb,
lib/jeventmachine.rb,
lib/em/process_watch.rb,
lib/em/protocols/stomp.rb,
lib/em/protocols/socks4.rb,
lib/em/protocols/tcptest.rb,
lib/em/threaded_resource.rb,
lib/em/protocols/memcache.rb,
lib/em/protocols/saslauth.rb,
lib/em/protocols/linetext2.rb,
lib/em/protocols/postgres3.rb,
lib/em/protocols/httpclient.rb,
lib/em/protocols/smtpclient.rb,
lib/em/protocols/smtpserver.rb,
lib/em/protocols/httpclient2.rb,
lib/em/protocols/line_and_text.rb,
lib/em/protocols/line_protocol.rb,
lib/em/protocols/object_protocol.rb,
lib/em/protocols/header_and_content.rb,
ext/rubymain.cpp,
ext/fastfilereader/rubymain.cpp
Overview
–
- Author
-
Francis Cianfrocca (gmail: blackhedd)
- Homepage
- Date
-
15 Nov 2006
See EventMachine and EventMachine::Connection for documentation and usage examples.
Copyright © 2006-07 by Francis Cianfrocca. All Rights Reserved. Gmail: blackhedd
This program is free software; you can redistribute it and/or modify it under the terms of either: 1) the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version; or 2) Ruby’s License.
See the file COPYING for complete licensing information.
Defined Under Namespace
Modules: DNS, Deferrable, Protocols, UuidGenerator Classes: Channel, Completion, Connection, ConnectionError, ConnectionNotBound, DatagramObject, DefaultDeferrable, DeferrableChildProcess, Error, EvmaKeyboard, EvmaTCPClient, EvmaTCPServer, EvmaUDPSocket, EvmaUNIXClient, EvmaUNIXServer, FastFileReader, FileNotFoundException, FileStreamer, FileWatch, Iterator, JEM, LoopbreakReader, PeriodicTimer, Pool, ProcessWatch, Queue, Reactor, Selectable, SpawnedProcess, StreamObject, SystemCmd, ThreadedResource, TickLoop, Timer, UnknownTimerFired, Unsupported, YieldBlockFromSpawnedProcess
Constant Summary collapse
- VERSION =
"1.2.0.1"
- TimerFired =
Connection states
INT2NUM(EM_TIMER_FIRED )
- ConnectionData =
INT2NUM(EM_CONNECTION_READ )
- ConnectionUnbound =
INT2NUM(EM_CONNECTION_UNBOUND )
- ConnectionAccepted =
INT2NUM(EM_CONNECTION_ACCEPTED )
- ConnectionCompleted =
INT2NUM(EM_CONNECTION_COMPLETED )
- LoopbreakSignalled =
INT2NUM(EM_LOOPBREAK_SIGNAL )
- ERRNOS =
System errnos
Errno::constants.grep(/^E/).inject(Hash.new(:unknown)) { |hash, name| errno = Errno.__send__(:const_get, name) hash[errno::Errno] = errno hash }
- P =
Alias for Protocols
EventMachine::Protocols
- ConnectionNotifyReadable =
INT2NUM(EM_CONNECTION_NOTIFY_READABLE)
- ConnectionNotifyWritable =
INT2NUM(EM_CONNECTION_NOTIFY_WRITABLE)
- SslHandshakeCompleted =
INT2NUM(EM_SSL_HANDSHAKE_COMPLETED )
- EM_PROTO_SSLv2 =
SSL Protocols
INT2NUM(EM_PROTO_SSLv2 )
- EM_PROTO_SSLv3 =
INT2NUM(EM_PROTO_SSLv3 )
- EM_PROTO_TLSv1 =
INT2NUM(EM_PROTO_TLSv1 )
- EM_PROTO_TLSv1_1 =
INT2NUM(EM_PROTO_TLSv1_1)
- EM_PROTO_TLSv1_2 =
INT2NUM(EM_PROTO_TLSv1_2)
Class Attribute Summary collapse
-
.reactor_thread ⇒ Thread
readonly
Exposed to allow joining on the thread, when run in a multithreaded environment.
- .threadpool ⇒ Object readonly
-
.threadpool_size ⇒ Number
Size of the EventMachine.defer threadpool (defaults to 20).
Class Method Summary collapse
- ._get_outbound_data_size(sig) ⇒ Object
- ._open_file_for_writing(filename, handler = nil) ⇒ Object
-
.add_oneshot_timer(interval) ⇒ Object
Changed 04Oct06: intervals from the caller are now in milliseconds, but our native-ruby processor still wants them in seconds.
-
.add_periodic_timer(*args, &block) ⇒ Object
Adds a periodic timer to the event loop.
-
.add_shutdown_hook(&block) ⇒ Object
Adds a block to call as the reactor is shutting down.
-
.add_timer(*args, &block) ⇒ Object
Adds a one-shot timer to the event loop.
-
.attach(io, handler = nil, *args, &blk) ⇒ Object
Attaches an IO object or file descriptor to the eventloop as a regular connection.
- .attach_fd(fileno, watch_mode) ⇒ Object
- .attach_io(io, watch_mode, handler = nil, *args) ⇒ Object
-
.attach_server(sock, handler = nil, *args, &block) ⇒ Object
Attach to an existing socket’s file descriptor.
-
.bind_connect(bind_addr, bind_port, server, port = nil, handler = nil, *args) ⇒ Object
This method is like EventMachine.connect, but allows for a local address/port to bind the connection to.
- .bind_connect_server(bind_addr, bind_port, server, port) ⇒ Object
-
.Callback(object = nil, method = nil, &blk) ⇒ <#call>
Utility method for coercing arguments to an object that responds to :call.
-
.cancel_timer(timer_or_sig) ⇒ Object
Cancel a timer (can be a callback or an Timer instance).
-
.cleanup_machine ⇒ Object
Clean up Ruby space following a release_machine.
-
.close_connection(sig, after_writing) ⇒ Object
The extension version does NOT raise any kind of an error if an attempt is made to close a non-existent connection.
-
.connect(server, port = nil, handler = nil, *args, &blk) ⇒ Object
Initiates a TCP connection to a remote server and sets up event handling for the connection.
- .connect_server(server, port) ⇒ Object
-
.connect_unix_domain(socketname, *args, &blk) ⇒ Object
Make a connection to a Unix-domain socket.
- .connect_unix_server(chain) ⇒ Object
-
.connection_count ⇒ Integer
Returns the total number of connections (file descriptors) currently held by the reactor.
- .connection_paused?(sig) ⇒ Boolean
-
.defer(op = nil, callback = nil, errback = nil, &blk) ⇒ Object
EventMachine.defer is used for integrating blocking operations into EventMachine’s control flow.
-
.defers_finished? ⇒ Boolean
Returns
true
if all deferred actions are done executing and their callbacks have been fired. - .detach_fd(sig) ⇒ Object
-
.disable_proxy(from) ⇒ Object
Takes just one argument, a Connection that has proxying enabled via EventMachine.enable_proxy.
-
.enable_proxy(from, to, bufsize = 0, length = 0) ⇒ Object
This method allows for direct writing of incoming data back out to another descriptor, at the C++ level in the reactor.
-
.epoll ⇒ Object
This method is a harmless no-op in the pure-Ruby implementation.
-
.epoll=(val) ⇒ Object
Epoll is a no-op for Java.
- .epoll? ⇒ Boolean
-
.error_handler(cb = nil, &blk) ⇒ Object
Catch-all for errors raised during event loop callbacks.
- .event_callback(conn_binding, opcode, data) ⇒ Object
-
.fork_reactor(&block) ⇒ Object
Forks a new process, properly stops the reactor and then calls EventMachine.run inside of it again, passing your block.
- .get_connection_count ⇒ Object
- .get_max_timer_count ⇒ Object
-
.get_max_timers ⇒ Integer
Gets the current maximum number of allowed timers.
- .get_outbound_data_size(sig) ⇒ Object
- .get_peername(sig) ⇒ Object
- .get_sock_opt(signature, level, optname) ⇒ Object
- .get_sockname(sig) ⇒ Object
-
.heartbeat_interval ⇒ Integer
Retrieve the heartbeat interval.
-
.heartbeat_interval=(time) ⇒ Object
Set the heartbeat interval.
-
.initialize_event_machine ⇒ Object
class Connection < com.rubyeventmachine.Connection def associate_callback_target sig # No-op for the time being.
- .invoke_popen(cmd) ⇒ Object
- .is_notify_readable(sig) ⇒ Object
- .is_notify_writable(sig) ⇒ Object
- .klass_from_handler(klass = Connection, handler = nil, *args) ⇒ Object
- .kqueue ⇒ Object
- .kqueue=(val) ⇒ Object
- .kqueue? ⇒ Boolean
-
.library_type ⇒ Object
This is mostly useful for automated tests.
-
.next_tick(pr = nil, &block) ⇒ Object
Schedules a proc for execution immediately after the next “turn” through the reactor core.
-
.open_datagram_socket(address, port, handler = nil, *args) ⇒ Object
Used for UDP-based protocols.
-
.open_keyboard(handler = nil, *args) ⇒ Object
(Experimental).
-
.open_udp_socket(server, port) ⇒ Object
Currently a no-op for Java.
- .pause_connection(sig) ⇒ Object
-
.popen(cmd, handler = nil, *args) {|c| ... } ⇒ Object
Runs an external process.
-
.reactor_running? ⇒ Boolean
Tells you whether the EventMachine reactor loop is currently running.
-
.reactor_thread? ⇒ Boolean
True if the calling thread is the same thread as the reactor.
- .read_keyboard ⇒ Object
-
.reconnect(server, port, handler) ⇒ Object
Connect to a given host/port and re-use the provided Connection instance.
- .release_machine ⇒ Object
- .resume_connection(sig) ⇒ Object
-
.run(blk = nil, tail = nil, &block) ⇒ Object
Initializes and runs an event loop.
-
.run_block(&block) ⇒ Object
Sugars a common use case.
-
.run_deferred_callbacks ⇒ Object
The is the responder for the loopback-signalled event.
- .run_machine ⇒ Object
-
.schedule(*a, &b) ⇒ Object
Runs the given callback on the reactor thread, or immediately if called from the reactor thread.
- .send_data(sig, data, length) ⇒ Object
-
.send_datagram(sig, data, length, address, port) ⇒ Object
This is currently only for UDP! We need to make it work with unix-domain sockets as well.
- .send_file_data(sig, filename) ⇒ Object
- .set_comm_inactivity_timeout(sig, interval) ⇒ Object
-
.set_descriptor_table_size(n_descriptors = nil) ⇒ Integer
Sets the maximum number of file or socket descriptors that your process may open.
-
.set_effective_user(username) ⇒ Object
A wrapper over the setuid system call.
- .set_heartbeat_interval(val) ⇒ Object
-
.set_max_timer_count(num) ⇒ Object
This method is a harmless no-op in pure Ruby, which doesn’t have a built-in limit on the number of available timers.
-
.set_max_timers(ct) ⇒ Object
Sets the maximum number of timers and periodic timers that may be outstanding at any given time.
- .set_notify_readable(sig, mode) ⇒ Object
- .set_notify_writable(sig, mode) ⇒ Object
- .set_pending_connect_timeout(sig, val) ⇒ Object
-
.set_quantum(mills) ⇒ Object
For advanced users.
-
.set_rlimit_nofile(n_descriptors) ⇒ Object
This method is a no-op in the pure-Ruby implementation.
- .set_sock_opt(signature, level, optname, optval) ⇒ Object
-
.set_timer_quantum(q) ⇒ Object
Sets reactor quantum in milliseconds.
- .set_tls_parms(sig, params) ⇒ Object
- .signal_loopbreak ⇒ Object
-
.spawn(&block) ⇒ Object
Spawn an erlang-style process.
- .spawn_threadpool ⇒ Object
-
.ssl? ⇒ Boolean
This method is not implemented for pure-Ruby implementation.
-
.start_server(server, port = nil, handler = nil, *args, &block) ⇒ Object
Initiates a TCP server (socket acceptor) on the specified IP address and port.
- .start_tcp_server(server, port) ⇒ Object
- .start_tls(sig) ⇒ Object
-
.start_unix_domain_server(filename, *args, &block) ⇒ Object
Start a Unix-domain server.
- .start_unix_server(filename) ⇒ Object
- .stop ⇒ Object
-
.stop_event_loop ⇒ Object
Causes the processing loop to stop executing, which will cause all open connections and accepting servers to be run down and closed.
-
.stop_server(signature) ⇒ Object
Stop a TCP server socket that was started with EventMachine.start_server.
- .stop_tcp_server(sig) ⇒ Object
- .stopping? ⇒ Boolean
-
.system(cmd, *args, &cb) ⇒ Object
EM::system is a simple wrapper for EM::popen.
-
.tick_loop(*a, &b) ⇒ Object
Creates and immediately starts an EventMachine::TickLoop.
-
.watch(io, handler = nil, *args, &blk) ⇒ Object
EventMachine.watch registers a given file descriptor or IO object with the eventloop.
-
.watch_file(filename, handler = nil, *args) ⇒ Object
EventMachine’s file monitoring API.
-
.watch_process(pid, handler = nil, *args) ⇒ Object
EventMachine’s process monitoring API.
- .yield(&block) ⇒ Object
- .yield_and_notify(&block) ⇒ Object
Class Attribute Details
.reactor_thread ⇒ Thread (readonly)
Exposed to allow joining on the thread, when run in a multithreaded environment. Performing other actions on the thread has undefined semantics (read: a dangerous endevor).
79 80 81 |
# File 'lib/eventmachine.rb', line 79 def reactor_thread @reactor_thread end |
.threadpool ⇒ Object (readonly)
1104 1105 1106 |
# File 'lib/eventmachine.rb', line 1104 def threadpool @threadpool end |
.threadpool_size ⇒ Number
Size of the EventMachine.defer threadpool (defaults to 20)
1108 1109 1110 |
# File 'lib/eventmachine.rb', line 1108 def threadpool_size @threadpool_size end |
Class Method Details
._get_outbound_data_size(sig) ⇒ Object
280 281 282 |
# File 'lib/jeventmachine.rb', line 280 def self._get_outbound_data_size(sig) @em.getOutboundDataSize(sig) end |
._open_file_for_writing(filename, handler = nil) ⇒ Object
1546 1547 1548 1549 1550 1551 1552 1553 1554 |
# File 'lib/eventmachine.rb', line 1546 def self._open_file_for_writing filename, handler=nil klass = klass_from_handler(Connection, handler) s = _write_file filename c = klass.new s @conns[s] = c block_given? and yield c c end |
.add_oneshot_timer(interval) ⇒ Object
Changed 04Oct06: intervals from the caller are now in milliseconds, but our native-ruby processor still wants them in seconds.
56 57 58 |
# File 'lib/em/pure_ruby.rb', line 56 def add_oneshot_timer interval Reactor.instance.install_oneshot_timer(interval / 1000) end |
.add_periodic_timer(*args, &block) ⇒ Object
Adds a periodic timer to the event loop. It takes the same parameters as the one-shot timer method, add_timer. This method schedules execution of the given block repeatedly, at intervals of time *at least* as great as the number of seconds given in the first parameter to the call.
350 351 352 353 354 355 |
# File 'lib/eventmachine.rb', line 350 def self.add_periodic_timer *args, &block interval = args.shift code = args.shift || block EventMachine::PeriodicTimer.new(interval, code) end |
.add_shutdown_hook(&block) ⇒ Object
Adds a block to call as the reactor is shutting down.
These callbacks are called in the reverse order to which they are added.
290 291 292 |
# File 'lib/eventmachine.rb', line 290 def self.add_shutdown_hook &block @tails << block end |
.add_timer(*args, &block) ⇒ Object
Adds a one-shot timer to the event loop. Call it with one or two parameters. The first parameters is a delay-time expressed in seconds (not milliseconds). The second parameter, if present, must be an object that responds to :call. If 2nd parameter is not given, then you can also simply pass a block to the method call.
This method may be called from the block passed to run or from any callback method. It schedules execution of the proc or block passed to it, after the passage of an interval of time equal to *at least* the number of seconds specified in the first parameter to the call.
add_timer is a non-blocking method. Callbacks can and will be called during the interval of time that the timer is in effect. There is no built-in limit to the number of timers that can be outstanding at any given time.
322 323 324 325 326 327 328 329 330 331 |
# File 'lib/eventmachine.rb', line 322 def self.add_timer *args, &block interval = args.shift code = args.shift || block if code # check too many timers! s = add_oneshot_timer((interval.to_f * 1000).to_i) @timers[s] = code s end end |
.attach(io, handler = nil, *args, &blk) ⇒ Object
Attaches an IO object or file descriptor to the eventloop as a regular connection. The file descriptor will be set as non-blocking, and EventMachine will process receive_data and send_data events on it as it would for any other connection.
To watch a fd instead, use watch, which will not alter the state of the socket and fire notify_readable and notify_writable events instead.
740 741 742 |
# File 'lib/eventmachine.rb', line 740 def EventMachine::attach io, handler=nil, *args, &blk attach_io io, false, handler, *args, &blk end |
.attach_fd(fileno, watch_mode) ⇒ Object
212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 |
# File 'lib/jeventmachine.rb', line 212 def self.attach_fd fileno, watch_mode # 3Aug09: We could pass in the actual SocketChannel, but then it would be modified (set as non-blocking), and # we would need some logic to make sure detach_fd below didn't clobber it. For now, we just always make a new # SocketChannel for the underlying file descriptor # if fileno.java_kind_of? SocketChannel # ch = fileno # ch.configureBlocking(false) # fileno = nil # elsif fileno.java_kind_of? java.nio.channels.Channel if fileno.java_kind_of? java.nio.channels.Channel field = fileno.getClass.getDeclaredField('fdVal') field.setAccessible(true) fileno = field.get(fileno) else raise ArgumentError, 'attach_fd requires Java Channel or POSIX fileno' unless fileno.is_a? Fixnum end if fileno == 0 raise "can't open STDIN as selectable in Java =(" elsif fileno.is_a? Fixnum # 8Aug09: The following code is specific to the sun jvm's SocketChannelImpl. Is there a cross-platform # way of implementing this? If so, also remember to update EventableSocketChannel#close and #cleanup fd = FileDescriptor.new fd.set_field 'fd', fileno ch = SocketChannel.open ch.configureBlocking(false) ch.kill ch.set_field 'fd', fd ch.set_field 'fdVal', fileno ch.set_field 'state', ch.get_field('ST_CONNECTED') end @em.attachChannel(ch,watch_mode) end |
.attach_io(io, watch_mode, handler = nil, *args) ⇒ Object
745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 |
# File 'lib/eventmachine.rb', line 745 def EventMachine::attach_io io, watch_mode, handler=nil, *args klass = klass_from_handler(Connection, handler, *args) if !watch_mode and klass.public_instance_methods.any?{|m| [:notify_readable, :notify_writable].include? m.to_sym } raise ArgumentError, "notify_readable/writable with EM.attach is not supported. Use EM.watch(io){ |c| c.notify_readable = true }" end if io.respond_to?(:fileno) # getDescriptorByFileno deprecated in JRuby 1.7.x, removed in JRuby 9000 if defined?(JRuby) && JRuby.runtime.respond_to?(:getDescriptorByFileno) fd = JRuby.runtime.getDescriptorByFileno(io.fileno).getChannel else fd = io.fileno end else fd = io end s = attach_fd fd, watch_mode c = klass.new s, *args c.instance_variable_set(:@io, io) c.instance_variable_set(:@watch_mode, watch_mode) c.instance_variable_set(:@fd, fd) @conns[s] = c block_given? and yield c c end |
.attach_server(sock, handler = nil, *args, &block) ⇒ Object
Attach to an existing socket’s file descriptor. The socket may have been started with start_server.
540 541 542 543 544 545 546 |
# File 'lib/eventmachine.rb', line 540 def self.attach_server sock, handler=nil, *args, &block klass = klass_from_handler(Connection, handler, *args) sd = sock.respond_to?(:fileno) ? sock.fileno : sock s = attach_sd(sd) @acceptors[s] = [klass,args,block,sock] s end |
.bind_connect(bind_addr, bind_port, server, port = nil, handler = nil, *args) ⇒ Object
This method is like connect, but allows for a local address/port to bind the connection to.
660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 |
# File 'lib/eventmachine.rb', line 660 def self.bind_connect bind_addr, bind_port, server, port=nil, handler=nil, *args begin port = Integer(port) rescue ArgumentError, TypeError # there was no port, so server must be a unix domain socket # the port argument is actually the handler, and the handler is one of the args args.unshift handler if handler handler = port port = nil end if port klass = klass_from_handler(Connection, handler, *args) s = if port if bind_addr bind_connect_server bind_addr, bind_port.to_i, server, port else connect_server server, port end else connect_unix_server server end c = klass.new s, *args @conns[s] = c block_given? and yield c c end |
.bind_connect_server(bind_addr, bind_port, server, port) ⇒ Object
85 86 87 |
# File 'lib/em/pure_ruby.rb', line 85 def bind_connect_server bind_addr, bind_port, host, port EvmaTCPClient.connect(bind_addr, bind_port, host, port).uuid end |
.Callback(object, method) ⇒ <#call> .Callback(object) ⇒ <#call> .Callback(&block) ⇒ <#call>
Utility method for coercing arguments to an object that responds to :call. Accepts an object and a method name to send to, or a block, or an object that responds to :call.
47 48 49 50 51 52 53 54 55 56 57 |
# File 'lib/em/callback.rb', line 47 def self.Callback(object = nil, method = nil, &blk) if object && method lambda { |*args| object.__send__ method, *args } else if object.respond_to? :call object else blk || raise(ArgumentError) end # if end # if end |
.cancel_timer(timer_or_sig) ⇒ Object
Cancel a timer (can be a callback or an Timer instance).
362 363 364 365 366 367 368 |
# File 'lib/eventmachine.rb', line 362 def self.cancel_timer timer_or_sig if timer_or_sig.respond_to? :cancel timer_or_sig.cancel else @timers[timer_or_sig] = false if @timers.has_key?(timer_or_sig) end end |
.cleanup_machine ⇒ Object
Clean up Ruby space following a release_machine
260 261 262 263 264 265 266 267 268 269 270 271 272 |
# File 'lib/eventmachine.rb', line 260 def self.cleanup_machine if @threadpool && !@threadpool.empty? # Tell the threads to stop @threadpool.each { |t| t.exit } # Join the threads or bump the stragglers one more time @threadpool.each { |t| t.join 0.01 || t.exit } end @threadpool = nil @threadqueue = nil @resultqueue = nil @all_threads_spawned = false @next_tick_queue = [] end |
.close_connection(sig, after_writing) ⇒ Object
The extension version does NOT raise any kind of an error if an attempt is made to close a non-existent connection. Not sure whether we should. For now, we’ll raise an error here in that case.
100 101 102 103 |
# File 'lib/em/pure_ruby.rb', line 100 def close_connection target, after_writing selectable = Reactor.instance.get_selectable( target ) or raise "unknown close_connection target" selectable.schedule_close after_writing end |
.connect(server, port = nil, handler = nil, *args, &blk) ⇒ Object
Initiates a TCP connection to a remote server and sets up event handling for the connection. connect requires event loop to be running (see run).
connect takes the IP address (or hostname) and port of the remote server you want to connect to. It also takes an optional handler (a module or a subclass of Connection) which you must define, that contains the callbacks that will be invoked by the event loop on behalf of the connection.
Learn more about connection lifecycle callbacks in the EventMachine tutorial and Connection lifecycle guide.
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 |
# File 'lib/eventmachine.rb', line 630 def self.connect server, port=nil, handler=nil, *args, &blk # EventMachine::connect initiates a TCP connection to a remote # server and sets up event-handling for the connection. # It internally creates an object that should not be handled # by the caller. HOWEVER, it's often convenient to get the # object to set up interfacing to other objects in the system. # We return the newly-created anonymous-class object to the caller. # It's expected that a considerable amount of code will depend # on this behavior, so don't change it. # # Ok, added support for a user-defined block, 13Apr06. # This leads us to an interesting choice because of the # presence of the post_init call, which happens in the # initialize method of the new object. We call the user's # block and pass the new object to it. This is a great # way to do protocol-specific initiation. It happens # AFTER post_init has been called on the object, which I # certainly hope is the right choice. # Don't change this lightly, because accepted connections # are different from connected ones and we don't want # to have them behave differently with respect to post_init # if at all possible. bind_connect nil, nil, server, port, handler, *args, &blk end |
.connect_server(server, port) ⇒ Object
80 81 82 |
# File 'lib/em/pure_ruby.rb', line 80 def connect_server host, port bind_connect_server nil, nil, host, port end |
.connect_unix_domain(socketname, *args, &blk) ⇒ Object
UNIX sockets, as the name suggests, are not available on Microsoft Windows.
Make a connection to a Unix-domain socket. This method is simply an alias for connect, which can connect to both TCP and Unix-domain sockets. Make sure that your process has sufficient permissions to open the socket it is given.
812 813 814 |
# File 'lib/eventmachine.rb', line 812 def self.connect_unix_domain socketname, *args, &blk connect socketname, *args, &blk end |
.connect_unix_server(chain) ⇒ Object
124 125 126 |
# File 'lib/em/pure_ruby.rb', line 124 def connect_unix_server chain EvmaUNIXClient.connect(chain).uuid end |
.connection_count ⇒ Integer
Returns the total number of connections (file descriptors) currently held by the reactor. Note that a tick must pass after the ‘initiation’ of a connection for this number to increment. It’s usually accurate, but don’t rely on the exact precision of this number unless you really know EM internals.
954 955 956 |
# File 'lib/eventmachine.rb', line 954 def self.connection_count self.get_connection_count end |
.connection_paused?(sig) ⇒ Boolean
277 278 279 |
# File 'lib/jeventmachine.rb', line 277 def self.connection_paused?(sig) @em.isConnectionPaused(sig) end |
.defer(op = nil, callback = nil, errback = nil, &blk) ⇒ Object
EventMachine.defer is used for integrating blocking operations into EventMachine’s control flow. The action of defer is to take the block specified in the first parameter (the “operation”) and schedule it for asynchronous execution on an internal thread pool maintained by EventMachine. When the operation completes, it will pass the result computed by the block (if any) back to the EventMachine reactor. Then, EventMachine calls the block specified in the second parameter to defer (the “callback”), as part of its normal event handling loop. The result computed by the operation block is passed as a parameter to the callback. You may omit the callback parameter if you don’t need to execute any code after the operation completes. If the operation raises an unhandled exception, the exception will be passed to the third parameter to defer (the “errback”), as part of its normal event handling loop. If no errback is provided, the exception will be allowed to blow through to the main thread immediately.
## Caveats ##
Note carefully that the code in your deferred operation will be executed on a separate thread from the main EventMachine processing and all other Ruby threads that may exist in your program. Also, multiple deferred operations may be running at once! Therefore, you are responsible for ensuring that your operation code is threadsafe.
Don’t write a deferred operation that will block forever. If so, the current implementation will not detect the problem, and the thread will never be returned to the pool. EventMachine limits the number of threads in its pool, so if you do this enough times, your subsequent deferred operations won’t get a chance to run.
The threads within the EventMachine’s thread pool have abort_on_exception set to true. As a result, if an unhandled exception is raised by the deferred operation and an errback is not provided, it will blow through to the main thread immediately. If the main thread is within an indiscriminate rescue block at that time, the exception could be handled improperly by the main thread.
1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 |
# File 'lib/eventmachine.rb', line 1042 def self.defer op = nil, callback = nil, errback = nil, &blk # OBSERVE that #next_tick hacks into this mechanism, so don't make any changes here # without syncing there. # # Running with $VERBOSE set to true gives a warning unless all ivars are defined when # they appear in rvalues. But we DON'T ever want to initialize @threadqueue unless we # need it, because the Ruby threads are so heavyweight. We end up with this bizarre # way of initializing @threadqueue because EventMachine is a Module, not a Class, and # has no constructor. unless @threadpool @threadpool = [] @threadqueue = ::Queue.new @resultqueue = ::Queue.new spawn_threadpool end @threadqueue << [op||blk,callback,errback] end |
.defers_finished? ⇒ Boolean
Returns true
if all deferred actions are done executing and their callbacks have been fired.
1094 1095 1096 1097 1098 1099 1100 |
# File 'lib/eventmachine.rb', line 1094 def self.defers_finished? return false if @threadpool and !@all_threads_spawned return false if @threadqueue and not @threadqueue.empty? return false if @resultqueue and not @resultqueue.empty? return false if @threadpool and @threadqueue.num_waiting != @threadpool.size return true end |
.detach_fd(sig) ⇒ Object
248 249 250 251 252 |
# File 'lib/jeventmachine.rb', line 248 def self.detach_fd sig if ch = @em.detachChannel(sig) ch.get_field 'fdVal' end end |
.disable_proxy(from) ⇒ Object
Takes just one argument, a Connection that has proxying enabled via enable_proxy. Calling this method will remove that functionality and your connection will begin receiving data via EventMachine::Connection#receive_data again.
1439 1440 1441 |
# File 'lib/eventmachine.rb', line 1439 def self.disable_proxy(from) EM::stop_proxy(from.signature) end |
.enable_proxy(from, to, bufsize = 0, length = 0) ⇒ Object
This method allows for direct writing of incoming data back out to another descriptor, at the C++ level in the reactor. This is very efficient and especially useful for proxies where high performance is required. Propogating data from a server response all the way up to Ruby, and then back down to the reactor to be sent back to the client, is often unnecessary and incurs a significant performance decrease.
The two arguments are instance of Connection subclasses, ‘from’ and ‘to’. ‘from’ is the connection whose inbound data you want relayed back out. ‘to’ is the connection to write it to.
Once you call this method, the ‘from’ connection will no longer get receive_data callbacks from the reactor, except in the case that ‘to’ connection has already closed when attempting to write to it. You can see in the example, that proxy_target_unbound will be called when this occurs. After that, further incoming data will be passed into receive_data as normal.
Note also that this feature supports different types of descriptors: TCP, UDP, and pipes. You can relay data from one kind to another, for example, feed a pipe from a UDP stream.
1429 1430 1431 |
# File 'lib/eventmachine.rb', line 1429 def self.enable_proxy(from, to, bufsize=0, length=0) EM::start_proxy(from.signature, to.signature, bufsize, length) end |
.epoll ⇒ Object
This method is a harmless no-op in the pure-Ruby implementation. This is intended to ensure that user code behaves properly across different EM implementations.
163 164 |
# File 'lib/em/pure_ruby.rb', line 163 def epoll end |
.epoll=(val) ⇒ Object
Epoll is a no-op for Java. The latest Java versions run epoll when possible in NIO.
164 165 |
# File 'lib/jeventmachine.rb', line 164 def self.epoll= val end |
.epoll? ⇒ Boolean
170 171 172 |
# File 'lib/jeventmachine.rb', line 170 def self.epoll? false end |
.error_handler(cb = nil, &blk) ⇒ Object
Catch-all for errors raised during event loop callbacks.
1362 1363 1364 1365 1366 1367 1368 |
# File 'lib/eventmachine.rb', line 1362 def self.error_handler cb = nil, &blk if cb or blk @error_handler = cb || blk elsif instance_variable_defined? :@error_handler remove_instance_variable :@error_handler end end |
.event_callback(conn_binding, opcode, data) ⇒ Object
1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 |
# File 'lib/eventmachine.rb', line 1462 def self.event_callback conn_binding, opcode, data # # Changed 27Dec07: Eliminated the hookable error handling. # No one was using it, and it degraded performance significantly. # It's in original_event_callback, which is dead code. # # Changed 25Jul08: Added a partial solution to the problem of exceptions # raised in user-written event-handlers. If such exceptions are not caught, # we must cause the reactor to stop, and then re-raise the exception. # Otherwise, the reactor doesn't stop and it's left on the call stack. # This is partial because we only added it to #unbind, where it's critical # (to keep unbind handlers from being re-entered when a stopping reactor # runs down open connections). It should go on the other calls to user # code, but the performance impact may be too large. # if opcode == ConnectionUnbound if c = @conns.delete( conn_binding ) begin if c.original_method(:unbind).arity != 0 c.unbind(data == 0 ? nil : EventMachine::ERRNOS[data]) else c.unbind end # If this is an attached (but not watched) connection, close the underlying io object. if c.instance_variable_defined?(:@io) and !c.instance_variable_get(:@watch_mode) io = c.instance_variable_get(:@io) begin io.close rescue Errno::EBADF, IOError end end rescue Exception => e if stopping? @wrapped_exception = $! stop else raise e end end elsif c = @acceptors.delete( conn_binding ) # no-op else if $! # Bubble user generated errors. @wrapped_exception = $! EM.stop else raise ConnectionNotBound, "received ConnectionUnbound for an unknown signature: #{conn_binding}" end end elsif opcode == ConnectionAccepted accep,args,blk = @acceptors[conn_binding] raise NoHandlerForAcceptedConnection unless accep c = accep.new data, *args @conns[data] = c blk and blk.call(c) c # (needed?) ## # The remaining code is a fallback for the pure ruby and java reactors. # In the C++ reactor, these events are handled in the C event_callback() in rubymain.cpp elsif opcode == ConnectionCompleted c = @conns[conn_binding] or raise ConnectionNotBound, "received ConnectionCompleted for unknown signature: #{conn_binding}" c.connection_completed elsif opcode == TimerFired t = @timers.delete( data ) return if t == false # timer cancelled t or raise UnknownTimerFired, "timer data: #{data}" t.call elsif opcode == ConnectionData c = @conns[conn_binding] or raise ConnectionNotBound, "received data #{data} for unknown signature: #{conn_binding}" c.receive_data data elsif opcode == LoopbreakSignalled run_deferred_callbacks elsif opcode == ConnectionNotifyReadable c = @conns[conn_binding] or raise ConnectionNotBound c.notify_readable elsif opcode == ConnectionNotifyWritable c = @conns[conn_binding] or raise ConnectionNotBound c.notify_writable end end |
.fork_reactor(&block) ⇒ Object
Forks a new process, properly stops the reactor and then calls run inside of it again, passing your block.
242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 |
# File 'lib/eventmachine.rb', line 242 def self.fork_reactor &block # This implementation is subject to change, especially if we clean up the relationship # of EM#run to @reactor_running. # Original patch by Aman Gupta. # Kernel.fork do if reactor_running? stop_event_loop release_machine cleanup_machine @reactor_running = false @reactor_thread = nil end run block end end |
.get_connection_count ⇒ Object
267 268 269 |
# File 'lib/jeventmachine.rb', line 267 def self.get_connection_count @em.getConnectionCount end |
.get_max_timer_count ⇒ Object
194 195 196 197 |
# File 'lib/jeventmachine.rb', line 194 def self.get_max_timer_count # harmless no-op in Java. There's no built-in timer limit. @max_timer_count || 100_000 end |
.get_max_timers ⇒ Integer
Gets the current maximum number of allowed timers
923 924 925 |
# File 'lib/eventmachine.rb', line 923 def self.get_max_timers get_max_timer_count end |
.get_outbound_data_size(sig) ⇒ Object
211 212 213 214 |
# File 'lib/em/pure_ruby.rb', line 211 def get_outbound_data_size sig r = Reactor.instance.get_selectable( sig ) or raise "unknown get_outbound_data_size target" r.get_outbound_data_size end |
.get_peername(sig) ⇒ Object
134 135 136 137 |
# File 'lib/em/pure_ruby.rb', line 134 def get_peername sig selectable = Reactor.instance.get_selectable( sig ) or raise "unknown get_peername target" selectable.get_peername end |
.get_sock_opt(signature, level, optname) ⇒ Object
186 187 188 189 |
# File 'lib/em/pure_ruby.rb', line 186 def get_sock_opt signature, level, optname selectable = Reactor.instance.get_selectable( signature ) or raise "unknown get_peername target" selectable.getsockopt level, optname end |
.get_sockname(sig) ⇒ Object
206 207 208 209 210 |
# File 'lib/jeventmachine.rb', line 206 def self.get_sockname sig if sockName = @em.getSockName(sig) Socket.pack_sockaddr_in(*sockName) end end |
.heartbeat_interval ⇒ Integer
Retrieve the heartbeat interval. This is how often EventMachine will check for dead connections that have had an inactivity timeout set via EventMachine::Connection#set_comm_inactivity_timeout. Default is 2 seconds.
1448 1449 1450 |
# File 'lib/eventmachine.rb', line 1448 def self.heartbeat_interval EM::get_heartbeat_interval end |
.heartbeat_interval=(time) ⇒ Object
Set the heartbeat interval. This is how often EventMachine will check for dead connections that have had an inactivity timeout set via EventMachine::Connection#set_comm_inactivity_timeout. Takes a Numeric number of seconds. Default is 2.
1457 1458 1459 |
# File 'lib/eventmachine.rb', line 1457 def self.heartbeat_interval=(time) EM::set_heartbeat_interval time.to_f end |
.initialize_event_machine ⇒ Object
class Connection < com.rubyeventmachine.Connection
def associate_callback_target sig
# No-op for the time being.
end
end
49 50 51 |
# File 'lib/em/pure_ruby.rb', line 49 def initialize_event_machine Reactor.instance.initialize_for_run end |
.invoke_popen(cmd) ⇒ Object
182 183 184 185 |
# File 'lib/jeventmachine.rb', line 182 def self.invoke_popen cmd # TEMPORARILY unsupported until someone figures out how to do it. raise "unsupported on this platform" end |
.is_notify_readable(sig) ⇒ Object
261 262 263 |
# File 'lib/jeventmachine.rb', line 261 def self.is_notify_readable sig @em.isNotifyReadable(sig) end |
.is_notify_writable(sig) ⇒ Object
264 265 266 |
# File 'lib/jeventmachine.rb', line 264 def self.is_notify_writable sig @em.isNotifyWritable(sig) end |
.klass_from_handler(klass = Connection, handler = nil, *args) ⇒ Object
1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 |
# File 'lib/eventmachine.rb', line 1557 def self.klass_from_handler(klass = Connection, handler = nil, *args) klass = if handler and handler.is_a?(Class) raise ArgumentError, "must provide module or subclass of #{klass.name}" unless klass >= handler handler elsif handler if defined?(handler::EM_CONNECTION_CLASS) handler::EM_CONNECTION_CLASS else handler::const_set(:EM_CONNECTION_CLASS, Class.new(klass) {include handler}) end else klass end arity = klass.instance_method(:initialize).arity expected = arity >= 0 ? arity : -(arity + 1) if (arity >= 0 and args.size != expected) or (arity < 0 and args.size < expected) raise ArgumentError, "wrong number of arguments for #{klass}#initialize (#{args.size} for #{expected})" end klass end |
.kqueue ⇒ Object
166 167 |
# File 'lib/jeventmachine.rb', line 166 def self.kqueue end |
.kqueue=(val) ⇒ Object
168 169 |
# File 'lib/jeventmachine.rb', line 168 def self.kqueue= val end |
.kqueue? ⇒ Boolean
173 174 175 |
# File 'lib/jeventmachine.rb', line 173 def self.kqueue? false end |
.library_type ⇒ Object
This is mostly useful for automated tests. Return a distinctive symbol so the caller knows whether he’s dealing with an extension or with a pure-Ruby library.
44 45 46 |
# File 'lib/em/pure_ruby.rb', line 44 def library_type :pure_ruby end |
.next_tick(pr = nil, &block) ⇒ Object
Schedules a proc for execution immediately after the next “turn” through the reactor core. An advanced technique, this can be useful for improving memory management and/or application responsiveness, especially when scheduling large amounts of data for writing to a network connection.
This method takes either a single argument (which must be a callable object) or a block.
1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 |
# File 'lib/eventmachine.rb', line 1120 def self.next_tick pr=nil, &block # This works by adding to the @resultqueue that's used for #defer. # The general idea is that next_tick is used when we want to give the reactor a chance # to let other operations run, either to balance the load out more evenly, or to let # outbound network buffers drain, or both. So we probably do NOT want to block, and # we probably do NOT want to be spinning any threads. A program that uses next_tick # but not #defer shouldn't suffer the penalty of having Ruby threads running. They're # extremely expensive even if they're just sleeping. raise ArgumentError, "no proc or block given" unless ((pr && pr.respond_to?(:call)) or block) @next_tick_mutex.synchronize do @next_tick_queue << ( pr || block ) end signal_loopbreak if reactor_running? end |
.open_datagram_socket(address, port, handler = nil, *args) ⇒ Object
Used for UDP-based protocols. Its usage is similar to that of start_server.
This method will create a new UDP (datagram) socket and bind it to the address and port that you specify. The normal callbacks (see start_server) will be called as events of interest occur on the newly-created socket, but there are some differences in how they behave.
EventMachine::Connection#receive_data will be called when a datagram packet is received on the socket, but unlike TCP sockets, the message boundaries of the received data will be respected. In other words, if the remote peer sent you a datagram of a particular size, you may rely on EventMachine::Connection#receive_data to give you the exact data in the packet, with the original data length. Also observe that Connection#receive_data may be called with a zero-length data payload, since empty datagrams are permitted in UDP.
EventMachine::Connection#send_data is available with UDP packets as with TCP, but there is an important difference. Because UDP communications are connectionless, there is no implicit recipient for the packets you send. Ordinarily you must specify the recipient for each packet you send. However, EventMachine provides for the typical pattern of receiving a UDP datagram from a remote peer, performing some operation, and then sending one or more packets in response to the same remote peer. To support this model easily, just use EventMachine::Connection#send_data in the code that you supply for EventMachine::Connection#receive_data.
EventMachine will provide an implicit return address for any messages sent to EventMachine::Connection#send_data within the context of a EventMachine::Connection#receive_data callback, and your response will automatically go to the correct remote peer.
Observe that the port number that you supply to open_datagram_socket may be zero. In this case, EventMachine will create a UDP socket that is bound to an [ephemeral port](en.wikipedia.org/wiki/Ephemeral_port). This is not appropriate for servers that must publish a well-known port to which remote peers may send datagrams. But it can be useful for clients that send datagrams to other servers. If you do this, you will receive any responses from the remote servers through the normal EventMachine::Connection#receive_data callback. Observe that you will probably have issues with firewalls blocking the ephemeral port numbers, so this technique is most appropriate for LANs.
If you wish to send datagrams to arbitrary remote peers (not necessarily ones that have sent data to which you are responding), then see EventMachine::Connection#send_datagram.
DO NOT call send_data from a datagram socket outside of a EventMachine::Connection#receive_data method. Use EventMachine::Connection#send_datagram. If you do use EventMachine::Connection#send_data outside of a EventMachine::Connection#receive_data method, you’ll get a confusing error because there is no “peer,” as #send_data requires (inside of EventMachine::Connection#receive_data, EventMachine::Connection#send_data “fakes” the peer as described above).
871 872 873 874 875 876 877 878 879 880 881 882 |
# File 'lib/eventmachine.rb', line 871 def self.open_datagram_socket address, port, handler=nil, *args # Replaced the implementation on 01Oct06. Thanks to Tobias Gustafsson for pointing # out that this originally did not take a class but only a module. klass = klass_from_handler(Connection, handler, *args) s = open_udp_socket address, port.to_i c = klass.new s, *args @conns[s] = c block_given? and yield c c end |
.open_keyboard(handler = nil, *args) ⇒ Object
(Experimental)
1234 1235 1236 1237 1238 1239 1240 1241 1242 |
# File 'lib/eventmachine.rb', line 1234 def self.open_keyboard handler=nil, *args klass = klass_from_handler(Connection, handler, *args) s = read_keyboard c = klass.new s, *args @conns[s] = c block_given? and yield c c end |
.open_udp_socket(server, port) ⇒ Object
Currently a no-op for Java.
140 141 142 |
# File 'lib/em/pure_ruby.rb', line 140 def open_udp_socket host, port EvmaUDPSocket.create(host, port).uuid end |
.pause_connection(sig) ⇒ Object
271 272 273 |
# File 'lib/jeventmachine.rb', line 271 def self.pause_connection(sig) @em.pauseConnection(sig) end |
.popen(cmd, handler = nil, *args) {|c| ... } ⇒ Object
This method is not supported on Microsoft Windows
Runs an external process.
1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 |
# File 'lib/eventmachine.rb', line 1197 def self.popen cmd, handler=nil, *args # At this moment, it's only available on Unix. # Perhaps misnamed since the underlying function uses socketpair and is full-duplex. klass = klass_from_handler(Connection, handler, *args) w = case cmd when Array cmd when String Shellwords::shellwords( cmd ) end w.unshift( w.first ) if w.first s = invoke_popen( w ) c = klass.new s, *args @conns[s] = c yield(c) if block_given? c end |
.reactor_running? ⇒ Boolean
Tells you whether the EventMachine reactor loop is currently running.
Useful when writing libraries that want to run event-driven code, but may be running in programs that are already event-driven. In such cases, if reactor_running? returns false, your code can invoke run and run your application code inside the block passed to that method. If this method returns true, just execute your event-aware code.
1226 1227 1228 |
# File 'lib/eventmachine.rb', line 1226 def self.reactor_running? @reactor_running && Process.pid == @reactor_pid end |
.reactor_thread? ⇒ Boolean
Returns true if the calling thread is the same thread as the reactor.
226 227 228 |
# File 'lib/eventmachine.rb', line 226 def self.reactor_thread? Thread.current == @reactor_thread end |
.read_keyboard ⇒ Object
217 218 219 |
# File 'lib/em/pure_ruby.rb', line 217 def read_keyboard EvmaKeyboard.open.uuid end |
.reconnect(server, port, handler) ⇒ Object
Connect to a given host/port and re-use the provided Connection instance. Consider also EventMachine::Connection#reconnect.
780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 |
# File 'lib/eventmachine.rb', line 780 def self.reconnect server, port, handler # Observe, the test for already-connected FAILS if we call a reconnect inside post_init, # because we haven't set up the connection in @conns by that point. # RESIST THE TEMPTATION to "fix" this problem by redefining the behavior of post_init. # # Changed 22Nov06: if called on an already-connected handler, just return the # handler and do nothing more. Originally this condition raised an exception. # We may want to change it yet again and call the block, if any. raise "invalid handler" unless handler.respond_to?(:connection_completed) #raise "still connected" if @conns.has_key?(handler.signature) return handler if @conns.has_key?(handler.signature) s = if port connect_server server, port else connect_unix_server server end handler.signature = s @conns[s] = handler block_given? and yield handler handler end |
.release_machine ⇒ Object
66 67 |
# File 'lib/em/pure_ruby.rb', line 66 def release_machine end |
.resume_connection(sig) ⇒ Object
274 275 276 |
# File 'lib/jeventmachine.rb', line 274 def self.resume_connection(sig) @em.resumeConnection(sig) end |
.run(blk = nil, tail = nil, &block) ⇒ Object
This method blocks calling thread. If you need to start EventMachine event loop from a Web app running on a non event-driven server (Unicorn, Apache Passenger, Mongrel), do it in a separate thread like demonstrated in one of the examples.
Initializes and runs an event loop. This method only returns if code inside the block passed to this method calls stop_event_loop. The block is executed after initializing its internal event loop but before running the loop, therefore this block is the right place to call any code that needs event loop to run, for example, start_server, connect or similar methods of libraries that use EventMachine under the hood (like ‘EventMachine::HttpRequest.new` or `AMQP.start`).
Programs that are run for long periods of time (e.g. servers) usually start event loop by calling run, and let it run “forever”. It’s also possible to use run to make a single client-connection to a remote server, process the data flow from that single connection, and then call stop_event_loop to stop, in other words, to run event loop for a short period of time (necessary to complete some operation) and then shut it down.
Once event loop is running, it is perfectly possible to start multiple servers and clients simultaneously: content-aware proxies like [Proxymachine](github.com/mojombo/proxymachine) do just that.
## Using EventMachine with Ruby on Rails and other Web application frameworks ##
Standalone applications often run event loop on the main thread, thus blocking for their entire lifespan. In case of Web applications, if you are running an EventMachine-based app server such as [Thin](code.macournoyer.com/thin/) or [Goliath](github.com/postrank-labs/goliath/), they start event loop for you. Servers like Unicorn, Apache Passenger or Mongrel occupy main Ruby thread to serve HTTP(S) requests. This means that calling run on the same thread is not an option (it will result in Web server never binding to the socket). In that case, start event loop in a separate thread as demonstrated below.
149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 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 210 |
# File 'lib/eventmachine.rb', line 149 def self.run blk=nil, tail=nil, &block # Obsoleted the use_threads mechanism. # 25Nov06: Added the begin/ensure block. We need to be sure that release_machine # gets called even if an exception gets thrown within any of the user code # that the event loop runs. The best way to see this is to run a unit # test with two functions, each of which calls {EventMachine.run} and each of # which throws something inside of #run. Without the ensure, the second test # will start without release_machine being called and will immediately throw # if @reactor_running and @reactor_pid != Process.pid # Reactor was started in a different parent, meaning we have forked. # Clean up reactor state so a new reactor boots up in this child. stop_event_loop release_machine cleanup_machine @reactor_running = false end tail and @tails.unshift(tail) if reactor_running? (b = blk || block) and b.call # next_tick(b) else @conns = {} @acceptors = {} @timers = {} @wrapped_exception = nil @next_tick_queue ||= [] @tails ||= [] begin @reactor_pid = Process.pid @reactor_running = true initialize_event_machine (b = blk || block) and add_timer(0, b) if @next_tick_queue && !@next_tick_queue.empty? add_timer(0) { signal_loopbreak } end @reactor_thread = Thread.current # Rubinius needs to come back into "Ruby space" for GC to work, # so we'll crank the machine here. if defined?(RUBY_ENGINE) && RUBY_ENGINE == "rbx" while run_machine_once; end else run_machine end ensure until @tails.empty? @tails.pop.call end release_machine cleanup_machine @reactor_running = false @reactor_thread = nil end raise @wrapped_exception if @wrapped_exception end end |
.run_block(&block) ⇒ Object
217 218 219 220 221 222 223 |
# File 'lib/eventmachine.rb', line 217 def self.run_block &block pr = proc { block.call EventMachine::stop } run(&pr) end |
.run_deferred_callbacks ⇒ Object
The is the responder for the loopback-signalled event. It can be fired either by code running on a separate thread (defer) or on the main thread (next_tick). It will often happen that a next_tick handler will reschedule itself. We consume a copy of the tick queue so that tick events scheduled by tick events have to wait for the next pass through the reactor core.
966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 |
# File 'lib/eventmachine.rb', line 966 def self.run_deferred_callbacks until (@resultqueue ||= []).empty? result,cback = @resultqueue.pop cback.call result if cback end # Capture the size at the start of this tick... size = @next_tick_mutex.synchronize { @next_tick_queue.size } size.times do |i| callback = @next_tick_mutex.synchronize { @next_tick_queue.shift } begin callback.call rescue exception_raised = true raise ensure # This is a little nasty. The problem is, if an exception occurs during # the callback, then we need to send a signal to the reactor to actually # do some work during the next_tick. The only mechanism we have from the # ruby side is next_tick itself, although ideally, we'd just drop a byte # on the loopback descriptor. EM.next_tick {} if exception_raised end end end |
.run_machine ⇒ Object
61 62 63 |
# File 'lib/em/pure_ruby.rb', line 61 def run_machine Reactor.instance.run end |
.schedule(*a, &b) ⇒ Object
Runs the given callback on the reactor thread, or immediately if called from the reactor thread. Accepts the same arguments as Callback
232 233 234 235 236 237 238 239 |
# File 'lib/eventmachine.rb', line 232 def self.schedule(*a, &b) cb = Callback(*a, &b) if reactor_running? && reactor_thread? cb.call else next_tick { cb.call } end end |
.send_data(sig, data, length) ⇒ Object
90 91 92 93 |
# File 'lib/em/pure_ruby.rb', line 90 def send_data target, data, datalength selectable = Reactor.instance.get_selectable( target ) or raise "unknown send_data target" selectable.send_data data end |
.send_datagram(sig, data, length, address, port) ⇒ Object
This is currently only for UDP! We need to make it work with unix-domain sockets as well.
147 148 149 150 |
# File 'lib/em/pure_ruby.rb', line 147 def send_datagram target, data, datalength, host, port selectable = Reactor.instance.get_selectable( target ) or raise "unknown send_data target" selectable.send_datagram data, Socket::pack_sockaddr_in(port, host) end |
.send_file_data(sig, filename) ⇒ Object
198 199 200 201 202 203 204 205 206 207 208 |
# File 'lib/em/pure_ruby.rb', line 198 def send_file_data sig, filename sz = File.size(filename) raise "file too large" if sz > 32*1024 data = begin File.read filename rescue "" end send_data sig, data, data.length end |
.set_comm_inactivity_timeout(sig, interval) ⇒ Object
222 223 224 225 |
# File 'lib/em/pure_ruby.rb', line 222 def set_comm_inactivity_timeout sig, tm r = Reactor.instance.get_selectable( sig ) or raise "unknown set_comm_inactivity_timeout target" r.set_inactivity_timeout tm end |
.set_descriptor_table_size(n_descriptors = nil) ⇒ Integer
Sets the maximum number of file or socket descriptors that your process may open. If you call this method with no arguments, it will simply return the current size of the descriptor table without attempting to change it.
The new limit on open descriptors only applies to sockets and other descriptors that belong to EventMachine. It has **no effect** on the number of descriptors you can create in ordinary Ruby code.
Not available on all platforms. Increasing the number of descriptors beyond its default limit usually requires superuser privileges. (See set_effective_user for a way to drop superuser privileges while your program is running.)
1167 1168 1169 |
# File 'lib/eventmachine.rb', line 1167 def self.set_descriptor_table_size n_descriptors=nil EventMachine::set_rlimit_nofile n_descriptors end |
.set_effective_user(username) ⇒ Object
This method has no effective implementation on Windows or in the pure-Ruby implementation of EventMachine
A wrapper over the setuid system call. Particularly useful when opening a network server on a privileged port because you can use this call to drop privileges after opening the port. Also very useful after a call to set_descriptor_table_size, which generally requires that you start your process with root privileges.
This method is intended for use in enforcing security requirements, consequently it will throw a fatal error and end your program if it fails.
1148 1149 1150 |
# File 'lib/eventmachine.rb', line 1148 def self.set_effective_user username EventMachine::setuid_string username end |
.set_heartbeat_interval(val) ⇒ Object
146 147 |
# File 'lib/jeventmachine.rb', line 146 def self.set_heartbeat_interval val end |
.set_max_timer_count(num) ⇒ Object
This method is a harmless no-op in pure Ruby, which doesn’t have a built-in limit on the number of available timers.
182 183 |
# File 'lib/em/pure_ruby.rb', line 182 def set_max_timer_count n end |
.set_max_timers(ct) ⇒ Object
This method has to be used before event loop is started.
Sets the maximum number of timers and periodic timers that may be outstanding at any given time. You only need to call set_max_timers if you need more than the default number of timers, which on most platforms is 1000.
916 917 918 |
# File 'lib/eventmachine.rb', line 916 def self.set_max_timers ct set_max_timer_count ct end |
.set_notify_readable(sig, mode) ⇒ Object
254 255 256 |
# File 'lib/jeventmachine.rb', line 254 def self.set_notify_readable sig, mode @em.setNotifyReadable(sig, mode) end |
.set_notify_writable(sig, mode) ⇒ Object
257 258 259 |
# File 'lib/jeventmachine.rb', line 257 def self.set_notify_writable sig, mode @em.setNotifyWritable(sig, mode) end |
.set_pending_connect_timeout(sig, val) ⇒ Object
144 145 |
# File 'lib/jeventmachine.rb', line 144 def self.set_pending_connect_timeout sig, val end |
.set_quantum(mills) ⇒ Object
For advanced users. This function sets the default timer granularity, which by default is slightly smaller than 100 milliseconds. Call this function to set a higher or lower granularity. The function affects the behavior of add_timer and add_periodic_timer. Most applications will not need to call this function.
Avoid setting the quantum to very low values because that may reduce performance under some extreme conditions. We recommend that you not use values lower than 10.
This method only can be used if event loop is running.
901 902 903 |
# File 'lib/eventmachine.rb', line 901 def self.set_quantum mills set_timer_quantum mills.to_i end |
.set_rlimit_nofile(n_descriptors) ⇒ Object
This method is a no-op in the pure-Ruby implementation. We simply return Ruby’s built-in per-process file-descriptor limit.
175 176 177 |
# File 'lib/em/pure_ruby.rb', line 175 def set_rlimit_nofile n 1024 end |
.set_sock_opt(signature, level, optname, optval) ⇒ Object
192 193 194 195 |
# File 'lib/em/pure_ruby.rb', line 192 def set_sock_opt signature, level, optname, optval selectable = Reactor.instance.get_selectable( signature ) or raise "unknown get_peername target" selectable.setsockopt level, optname, optval end |
.set_timer_quantum(q) ⇒ Object
Sets reactor quantum in milliseconds. The underlying Reactor function wants a (possibly fractional) number of seconds.
156 157 158 |
# File 'lib/em/pure_ruby.rb', line 156 def set_timer_quantum interval Reactor.instance.set_timer_quantum(( 1.0 * interval) / 1000.0) end |
.set_tls_parms(sig, params) ⇒ Object
285 286 |
# File 'lib/jeventmachine.rb', line 285 def self.set_tls_parms(sig, params) end |
.signal_loopbreak ⇒ Object
129 130 131 |
# File 'lib/em/pure_ruby.rb', line 129 def signal_loopbreak Reactor.instance.signal_loopbreak end |
.spawn(&block) ⇒ Object
Spawn an erlang-style process
69 70 71 72 73 |
# File 'lib/em/spawnable.rb', line 69 def self.spawn &block s = SpawnedProcess.new s.set_receiver block s end |
.spawn_threadpool ⇒ Object
1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 |
# File 'lib/eventmachine.rb', line 1064 def self.spawn_threadpool until @threadpool.size == @threadpool_size.to_i thread = Thread.new do Thread.current.abort_on_exception = true while true begin op, cback, eback = *@threadqueue.pop rescue ThreadError $stderr.puts $!. break # Ruby 2.0 may fail at Queue.pop end begin result = op.call @resultqueue << [result, cback] rescue Exception => error raise error unless eback @resultqueue << [error, eback] end EventMachine.signal_loopbreak end end @threadpool << thread end @all_threads_spawned = true end |
.ssl? ⇒ Boolean
This method is not implemented for pure-Ruby implementation
168 169 170 |
# File 'lib/em/pure_ruby.rb', line 168 def ssl? false end |
.start_server(server, port = nil, handler = nil, *args, &block) ⇒ Object
Don’t forget that in order to bind to ports < 1024 on Linux, *BSD and Mac OS X your process must have superuser privileges.
Initiates a TCP server (socket acceptor) on the specified IP address and port.
The IP address must be valid on the machine where the program runs, and the process must be privileged enough to listen on the specified port (on Unix-like systems, superuser privileges are usually required to listen on any port lower than 1024). Only one listener may be running on any given address/port combination. start_server will fail if the given address and port are already listening on the machine, either because of a prior call to start_server or some unrelated process running on the machine. If start_server succeeds, the new network listener becomes active immediately and starts accepting connections from remote peers, and these connections generate callback events that are processed by the code specified in the handler parameter to start_server.
The optional handler which is passed to this method is the key to EventMachine’s ability to handle particular network protocols. The handler parameter passed to start_server must be a Ruby Module that you must define. When the network server that is started by start_server accepts a new connection, it instantiates a new object of an anonymous class that is inherited from Connection, *into which your handler module have been included*. Arguments passed into start_server after the class name are passed into the constructor during the instantiation.
Your handler module may override any of the methods in Connection, such as EventMachine::Connection#receive_data, in order to implement the specific behavior of the network protocol.
Callbacks invoked in response to network events always take place within the execution context of the object derived from Connection extended by your handler module. There is one object per connection, and all of the callbacks invoked for a particular connection take the form of instance methods called against the corresponding Connection object. Therefore, you are free to define whatever instance variables you wish, in order to contain the per-connection state required by the network protocol you are implementing.
start_server is usually called inside the block passed to run, but it can be called from any EventMachine callback. start_server will fail unless the EventMachine event loop is currently running (which is why it’s often called in the block suppled to run).
You may call start_server any number of times to start up network listeners on different address/port combinations. The servers will all run simultaneously. More interestingly, each individual call to start_server can specify a different handler module and thus implement a different network protocol from all the others.
516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 |
# File 'lib/eventmachine.rb', line 516 def self.start_server server, port=nil, handler=nil, *args, &block begin port = Integer(port) rescue ArgumentError, TypeError # there was no port, so server must be a unix domain socket # the port argument is actually the handler, and the handler is one of the args args.unshift handler if handler handler = port port = nil end if port klass = klass_from_handler(Connection, handler, *args) s = if port start_tcp_server server, port else start_unix_server server end @acceptors[s] = [klass,args,block] s end |
.start_tcp_server(server, port) ⇒ Object
106 107 108 109 |
# File 'lib/em/pure_ruby.rb', line 106 def start_tcp_server host, port (s = EvmaTCPServer.start_server host, port) or raise "no acceptor" s.uuid end |
.start_tls(sig) ⇒ Object
148 149 150 |
# File 'lib/jeventmachine.rb', line 148 def self.start_tls sig @em.startTls sig end |
.start_unix_domain_server(filename, *args, &block) ⇒ Object
Start a Unix-domain server.
Note that this is an alias for start_server, which can be used to start both TCP and Unix-domain servers.
560 561 562 |
# File 'lib/eventmachine.rb', line 560 def self.start_unix_domain_server filename, *args, &block start_server filename, *args, &block end |
.start_unix_server(filename) ⇒ Object
118 119 120 121 |
# File 'lib/em/pure_ruby.rb', line 118 def start_unix_server chain (s = EvmaUNIXServer.start_server chain) or raise "no acceptor" s.uuid end |
.stop ⇒ Object
75 76 77 |
# File 'lib/em/pure_ruby.rb', line 75 def stop Reactor.instance.stop end |
.stop_event_loop ⇒ Object
Causes the processing loop to stop executing, which will cause all open connections and accepting servers to be run down and closed. Connection termination callbacks added using add_shutdown_hook will be called as part of running this method.
When all of this processing is complete, the call to run which started the processing loop will return and program flow will resume from the statement following run call.
416 417 418 |
# File 'lib/eventmachine.rb', line 416 def self.stop_event_loop EventMachine::stop end |
.stop_server(signature) ⇒ Object
Stop a TCP server socket that was started with start_server.
550 551 552 |
# File 'lib/eventmachine.rb', line 550 def self.stop_server signature EventMachine::stop_tcp_server signature end |
.stop_tcp_server(sig) ⇒ Object
112 113 114 115 |
# File 'lib/em/pure_ruby.rb', line 112 def stop_tcp_server sig s = Reactor.instance.get_selectable(sig) s.schedule_close end |
.stopping? ⇒ Boolean
70 71 72 |
# File 'lib/em/pure_ruby.rb', line 70 def stopping? return Reactor.instance.stop_scheduled end |
.system(cmd, *args, &cb) ⇒ Object
EM::system is a simple wrapper for EM::popen. It is similar to Kernel::system, but requires a single string argument for the command and performs no shell expansion.
The block or proc passed to EM::system is called with two arguments: the output generated by the command, and a Process::Status that contains information about the command’s execution.
EM.run{
EM.system('ls'){ |output,status| puts output if status.exitstatus == 0 }
}
You can also supply an additional proc to send some data to the process:
EM.run{
EM.system('sh', proc{ |process|
process.send_data("echo hello\n")
process.send_data("exit\n")
}, proc{ |out,status|
puts(out)
})
}
Like EventMachine.popen, EventMachine.system currently does not work on windows. It returns the pid of the spawned process.
112 113 114 115 116 117 118 119 120 121 122 |
# File 'lib/em/processes.rb', line 112 def EventMachine::system cmd, *args, &cb cb ||= args.pop if args.last.is_a? Proc init = args.pop if args.last.is_a? Proc # merge remaining arguments into the command cmd = [cmd, *args] if args.any? EM.get_subprocess_pid(EM.popen(cmd, SystemCmd, cb) do |c| init[c] if init end.signature) end |
.tick_loop(*a, &b) ⇒ Object
Creates and immediately starts an EventMachine::TickLoop
3 4 5 |
# File 'lib/em/tick_loop.rb', line 3 def self.tick_loop(*a, &b) TickLoop.new(*a, &b).start end |
.watch(io, handler = nil, *args, &blk) ⇒ Object
watch registers a given file descriptor or IO object with the eventloop. The file descriptor will not be modified (it will remain blocking or non-blocking).
The eventloop can be used to process readable and writable events on the file descriptor, using EventMachine::Connection#notify_readable= and EventMachine::Connection#notify_writable=
EventMachine::Connection#notify_readable? and EventMachine::Connection#notify_writable? can be used to check what events are enabled on the connection.
To detach the file descriptor, use EventMachine::Connection#detach
730 731 732 |
# File 'lib/eventmachine.rb', line 730 def EventMachine::watch io, handler=nil, *args, &blk attach_io io, true, handler, *args, &blk end |
.watch_file(filename, handler = nil, *args) ⇒ Object
The ability to pick up on the new filename after a rename is not yet supported. Calling #path will always return the filename you originally used.
EventMachine’s file monitoring API. Currently supported are the following events on individual files, using inotify on Linux systems, and kqueue for *BSD and Mac OS X:
-
File modified (written to)
-
File moved/renamed
-
File deleted
EventMachine::watch_file takes a filename and a handler Module containing your custom callback methods. This will setup the low level monitoring on the specified file, and create a new EventMachine::FileWatch object with your Module mixed in. FileWatch is a subclass of Connection, so callbacks on this object work in the familiar way. The callbacks that will be fired by EventMachine are:
-
file_modified
-
file_moved
-
file_deleted
You can access the filename being monitored from within this object using EventMachine::FileWatch#path.
When a file is deleted, EventMachine::FileWatch#stop_watching will be called after your file_deleted callback, to clean up the underlying monitoring and remove EventMachine’s reference to the now-useless FileWatch instance. This will in turn call unbind, if you wish to use it.
The corresponding system-level Errno will be raised when attempting to monitor non-existent files, files with wrong permissions, or if an error occurs dealing with inotify/kqueue.
1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 |
# File 'lib/eventmachine.rb', line 1308 def self.watch_file(filename, handler=nil, *args) klass = klass_from_handler(FileWatch, handler, *args) s = EM::watch_filename(filename) c = klass.new s, *args # we have to set the path like this because of how Connection.new works c.instance_variable_set("@path", filename) @conns[s] = c block_given? and yield c c end |
.watch_process(pid, handler = nil, *args) ⇒ Object
EventMachine’s process monitoring API. On Mac OS X and *BSD this method is implemented using kqueue.
1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 |
# File 'lib/eventmachine.rb', line 1339 def self.watch_process(pid, handler=nil, *args) pid = pid.to_i klass = klass_from_handler(ProcessWatch, handler, *args) s = EM::watch_pid(pid) c = klass.new s, *args # we have to set the path like this because of how Connection.new works c.instance_variable_set("@pid", pid) @conns[s] = c block_given? and yield c c end |
.yield(&block) ⇒ Object
76 77 78 |
# File 'lib/em/spawnable.rb', line 76 def self.yield &block return YieldBlockFromSpawnedProcess.new( block, false ) end |
.yield_and_notify(&block) ⇒ Object
81 82 83 |
# File 'lib/em/spawnable.rb', line 81 def self.yield_and_notify &block return YieldBlockFromSpawnedProcess.new( block, true ) end |