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/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/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
Overview
--
Author:: Francis Cianfrocca (gmail: blackhedd) Homepage:: http://rubyeventmachine.com Date:: 15 Nov 2006
See EventMachine and EventMachine::Connection for documentation and usage examples.
Copyright (C) 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 Classes: Channel, Completion, Connection, DefaultDeferrable, DeferrableChildProcess, FileNotFoundException, FileStreamer, FileWatch, Iterator, PeriodicTimer, Pool, ProcessWatch, Queue, SpawnedProcess, ThreadedResource, TickLoop, Timer
Constant Summary collapse
Class Attribute Summary collapse
-
.reactor_thread ⇒ Thread
readonly
Exposed to allow joining on the thread, when run in a multithreaded environment.
-
.threadpool_size ⇒ Number
Size of the EventMachine.defer threadpool (defaults to 20).
Class Method Summary collapse
-
.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.
-
.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.
-
.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).
-
.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_unix_domain(socketname, *args, &blk) ⇒ Object
Make a connection to a Unix-domain socket.
-
.connection_count ⇒ Integer
Returns the total number of connections (file descriptors) currently held by the reactor.
-
.defer(op = nil, callback = 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.
-
.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.
-
.error_handler(cb = nil, &blk) ⇒ Object
Catch-all for errors raised during event loop callbacks.
-
.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_max_timers ⇒ Integer
Gets the current maximum number of allowed timers.
-
.heartbeat_interval ⇒ Integer
Retrieve the heartbeat interval.
-
.heartbeat_interval=(time) ⇒ Object
Set the heartbeat interval.
-
.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.
-
.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.
-
.reconnect(server, port, handler) ⇒ Object
Connect to a given host/port and re-use the provided Connection instance.
-
.run(blk = nil, tail = nil, &block) ⇒ Object
Initializes and runs an event loop.
-
.run_block(&block) ⇒ Object
Sugars a common use case.
-
.schedule(*a, &b) ⇒ Object
Runs the given callback on the reactor thread, or immediately if called from the reactor thread.
-
.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_max_timers(ct) ⇒ Object
Sets the maximum number of timers and periodic timers that may be outstanding at any given time.
-
.set_quantum(mills) ⇒ Object
For advanced users.
-
.spawn(&block) ⇒ Object
Spawn an erlang-style process.
-
.start_server(server, port = nil, handler = nil, *args, &block) ⇒ Object
Initiates a TCP server (socket acceptor) on the specified IP address and port.
-
.start_unix_domain_server(filename, *args, &block) ⇒ Object
Start a Unix-domain server.
-
.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.
-
.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.
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_size ⇒ Number
Size of the EventMachine.defer threadpool (defaults to 20)
1067 1068 1069 |
# File 'lib/eventmachine.rb', line 1067 def threadpool_size @threadpool_size end |
Class Method Details
.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.
348 349 350 351 352 353 |
# File 'lib/eventmachine.rb', line 348 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.
288 289 290 |
# File 'lib/eventmachine.rb', line 288 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.
320 321 322 323 324 325 326 327 328 329 |
# File 'lib/eventmachine.rb', line 320 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.
729 730 731 |
# File 'lib/eventmachine.rb', line 729 def EventMachine::attach io, handler=nil, *args, &blk attach_io io, false, handler, *args, &blk 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.
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 675 676 |
# File 'lib/eventmachine.rb', line 649 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 |
.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).
360 361 362 363 364 365 366 |
# File 'lib/eventmachine.rb', line 360 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 |
.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.
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 |
# File 'lib/eventmachine.rb', line 619 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_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.
796 797 798 |
# File 'lib/eventmachine.rb', line 796 def self.connect_unix_domain socketname, *args, &blk connect socketname, *args, &blk 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.
938 939 940 |
# File 'lib/eventmachine.rb', line 938 def self.connection_count self.get_connection_count end |
.defer(op = nil, callback = 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.
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.
1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 |
# File 'lib/eventmachine.rb', line 1011 def self.defer op = nil, callback = 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] end |
.defers_finished? ⇒ Boolean
Returns +true+ if all deferred actions are done executing and their callbacks have been fired.
1053 1054 1055 1056 1057 1058 1059 |
# File 'lib/eventmachine.rb', line 1053 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 |
.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.
1393 1394 1395 |
# File 'lib/eventmachine.rb', line 1393 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.
1383 1384 1385 |
# File 'lib/eventmachine.rb', line 1383 def self.enable_proxy(from, to, bufsize=0, length=0) EM::start_proxy(from.signature, to.signature, bufsize, length) end |
.error_handler(cb = nil, &blk) ⇒ Object
Catch-all for errors raised during event loop callbacks.
1316 1317 1318 1319 1320 1321 1322 |
# File 'lib/eventmachine.rb', line 1316 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 |
.fork_reactor(&block) ⇒ Object
Forks a new process, properly stops the reactor and then calls run inside of it again, passing your block.
257 258 259 260 261 262 263 264 265 266 267 268 269 270 |
# File 'lib/eventmachine.rb', line 257 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 self.reactor_running? self.stop_event_loop self.release_machine @reactor_running = false end self.run block end end |
.get_max_timers ⇒ Integer
Gets the current maximum number of allowed timers
907 908 909 |
# File 'lib/eventmachine.rb', line 907 def self.get_max_timers get_max_timer_count 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.
1402 1403 1404 |
# File 'lib/eventmachine.rb', line 1402 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.
1411 1412 1413 |
# File 'lib/eventmachine.rb', line 1411 def self.heartbeat_interval=(time) EM::set_heartbeat_interval time.to_f 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.
1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 |
# File 'lib/eventmachine.rb', line 1079 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. 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).
855 856 857 858 859 860 861 862 863 864 865 866 |
# File 'lib/eventmachine.rb', line 855 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 |
.popen(cmd, handler = nil, *args) {|c| ... } ⇒ Object
This method is not supported on Microsoft Windows
Runs an external process.
1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 |
# File 'lib/eventmachine.rb', line 1156 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 = Shellwords::shellwords( cmd ) 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.
1180 1181 1182 |
# File 'lib/eventmachine.rb', line 1180 def self.reactor_running? (@reactor_running || false) end |
.reactor_thread? ⇒ Boolean
Returns true if the calling thread is the same thread as the reactor.
241 242 243 |
# File 'lib/eventmachine.rb', line 241 def self.reactor_thread? Thread.current == @reactor_thread end |
.reconnect(server, port, handler) ⇒ Object
Connect to a given host/port and re-use the provided Connection instance. Consider also EventMachine::Connection#reconnect.
764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 |
# File 'lib/eventmachine.rb', line 764 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 |
.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 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 or 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.
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 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 |
# File 'lib/eventmachine.rb', line 151 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 @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 run_machine ensure until @tails.empty? @tails.pop.call end begin release_machine ensure if @threadpool @threadpool.each { |t| t.exit } @threadpool.each do |t| next unless t.alive? begin # Thread#kill! does not exist on 1.9 or rbx, and raises # NotImplemented on jruby t.kill! rescue NoMethodError, NotImplementedError t.kill # XXX t.join here? end end @threadqueue = nil @resultqueue = nil @threadpool = nil @all_threads_spawned = false end @next_tick_queue = [] end @reactor_running = false @reactor_thread = nil end raise @wrapped_exception if @wrapped_exception end end |
.run_block(&block) ⇒ Object
Sugars a common use case. Will pass the given block to #run, but will terminate the reactor loop and exit the function as soon as the code in the block completes. (Normally, run keeps running indefinitely, even after the block supplied to it finishes running, until user code calls stop)
232 233 234 235 236 237 238 |
# File 'lib/eventmachine.rb', line 232 def self.run_block &block pr = proc { block.call EventMachine::stop } run(&pr) 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
247 248 249 250 251 252 253 254 |
# File 'lib/eventmachine.rb', line 247 def self.schedule(*a, &b) cb = Callback(*a, &b) if reactor_running? && reactor_thread? cb.call else next_tick { cb.call } end 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.)
1126 1127 1128 |
# File 'lib/eventmachine.rb', line 1126 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.
1107 1108 1109 |
# File 'lib/eventmachine.rb', line 1107 def self.set_effective_user username EventMachine::setuid_string username 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.
900 901 902 |
# File 'lib/eventmachine.rb', line 900 def self.set_max_timers ct set_max_timer_count ct 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.
885 886 887 |
# File 'lib/eventmachine.rb', line 885 def self.set_quantum mills set_timer_quantum mills.to_i 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 |
.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.
514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 |
# File 'lib/eventmachine.rb', line 514 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_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.
549 550 551 |
# File 'lib/eventmachine.rb', line 549 def self.start_unix_domain_server filename, *args, &block start_server filename, *args, &block 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.
414 415 416 |
# File 'lib/eventmachine.rb', line 414 def self.stop_event_loop EventMachine::stop end |
.stop_server(signature) ⇒ Object
Stop a TCP server socket that was started with start_server.
539 540 541 |
# File 'lib/eventmachine.rb', line 539 def self.stop_server signature EventMachine::stop_tcp_server signature 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.map{|a|a.to_s.dump}).join(' ') 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
719 720 721 |
# File 'lib/eventmachine.rb', line 719 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.
1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 |
# File 'lib/eventmachine.rb', line 1262 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.
1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 |
# File 'lib/eventmachine.rb', line 1293 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 |