Changeset 723

Show
Ignore:
Timestamp:
06/21/08 08:20:07 (5 months ago)
Author:
raggi
Message:

Merge forward from trunk r712 to HEAD

Files:

Legend:

Unmodified
Added
Removed
Modified
Copied
Moved

  • branches/raggi/java/src/com/rubyeventmachine/EmReactor.java

    r668 r723  
    313313                sendData (sig, ByteBuffer.wrap(data.getBytes())); 
    314314                //(Connections.get(sig)).scheduleOutboundData( ByteBuffer.wrap(data.getBytes())); 
     315        } 
     316        public void setCommInactivityTimeout (String sig, long mills) { 
     317                (Connections.get(sig)).setCommInactivityTimeout (mills); 
    315318        } 
    316319         

  • branches/raggi/java/src/com/rubyeventmachine/EventableChannel.java

    r668 r723  
    5353         
    5454        public boolean writeOutboundData(); 
     55 
     56        public void setCommInactivityTimeout (long seconds); 
    5557} 

  • branches/raggi/java/src/com/rubyeventmachine/EventableDatagramChannel.java

    r668 r723  
    164164                return (bCloseScheduled && outboundQ.isEmpty()) ? false : true; 
    165165        } 
     166 
     167        public void setCommInactivityTimeout (long seconds) { 
     168                // TODO 
     169                System.out.println ("DATAGRAM: SET COMM INACTIVITY UNIMPLEMENTED " + seconds); 
     170        } 
    166171} 

  • branches/raggi/java/src/com/rubyeventmachine/EventableSocketChannel.java

    r668 r723  
    5454        LinkedList<ByteBuffer> outboundQ; 
    5555        boolean bCloseScheduled; 
     56        boolean bConnectPending; 
    5657         
    5758        SSLEngine sslEngine; 
    …… 
    6667                selector = sel; 
    6768                bCloseScheduled = false; 
     69                bConnectPending = false; 
    6870                outboundQ = new LinkedList<ByteBuffer>(); 
    6971                 
    …… 
    98100                                        outboundQ.addLast(bb); 
    99101                                } 
    100                                 channel.register(selector, SelectionKey.OP_WRITE | SelectionKey.OP_READ, this); 
     102                                channel.register(selector, SelectionKey.OP_WRITE | SelectionKey.OP_READ | (bConnectPending ? SelectionKey.OP_CONNECT : 0), this); 
    101103                        } 
    102104                } catch (ClosedChannelException e) { 
    …… 
    169171        public void setConnectPending() throws ClosedChannelException { 
    170172                channel.register(selector, SelectionKey.OP_CONNECT, this); 
     173                bConnectPending = true; 
    171174        } 
    172175         
    …… 
    183186                        return false; 
    184187                } 
    185                 channel.register(selector, SelectionKey.OP_READ, this); 
     188                bConnectPending = false; 
     189                channel.register(selector, SelectionKey.OP_READ | (outboundQ.isEmpty() ? 0 : SelectionKey.OP_WRITE), this); 
    186190                return true; 
    187191        } 
    188192         
    189193        public void scheduleClose (boolean afterWriting) { 
     194                // TODO: What the hell happens here if bConnectPending is set? 
    190195                if (!afterWriting) 
    191196                        outboundQ.clear(); 
    …… 
    232237                        return bb; 
    233238        } 
     239 
     240        public void setCommInactivityTimeout (long seconds) { 
     241                // TODO 
     242                System.out.println ("SOCKET: SET COMM INACTIVITY UNIMPLEMENTED " + seconds); 
     243        } 
    234244} 

  • branches/raggi/lib/jeventmachine.rb

    r712 r723  
    7676                @em.stopTcpServer sig 
    7777        end 
     78        def self.start_unix_server filename 
     79                # TEMPORARILY unsupported until someone figures out how to do it. 
     80                raise "unsupported on this platform" 
     81        end 
    7882        def self.send_data sig, data, length 
    7983                @em.sendData sig, data, length 
    …… 
    8791        def self.close_connection sig, after_writing 
    8892                @em.closeConnection sig, after_writing 
     93        end 
     94        def self.set_comm_inactivity_timeout sig, interval 
     95                @em.setCommInactivityTimeout sig, interval 
    8996        end 
    9097        def self.start_tls sig 
    …… 
    107114                @em.openUdpSocket server, port 
    108115        end 
     116        def self.invoke_popen cmd 
     117                # TEMPORARILY unsupported until someone figures out how to do it. 
     118                raise "unsupported on this platform" 
     119        end 
     120        def self.read_keyboard 
     121                # TEMPORARILY unsupported until someone figures out how to do it. 
     122                raise "temporarily unsupported on this platform" 
     123        end 
     124        def self.set_max_timer_count num 
     125                # harmless no-op in Java. There's no built-in timer limit. 
     126        end 
    109127        def self.library_type 
    110128                :java 

  • branches/raggi/tasks/tests.rake

    r708 r723  
    88  runner = Test::Unit::UI::Console::TestRunner 
    99 
    10   $eventmachine_library = libr 
     10  $eventmachine_library = ((RUBY_PLATFORM =~ /java/) ? :java : libr) 
    1111  $LOAD_PATH.unshift('tests') 
    1212  $stderr.puts "Checking for test cases:" #if t.verbose 

  • branches/raggi/tests/test_hc.rb

    r668 r723  
    3434# tests work. Native threads do strange things when you do I/O on them. 
    3535# 
     36# And it's even worse in Java, where I/O on native threads doesn't seem 
     37# to be reliable at all. 
     38# 
    3639 
    3740 
    3841class TestHeaderAndContentProtocol < Test::Unit::TestCase 
    3942 
    40     TestHost = "127.0.0.1" 
    41     TestPort = 8905 
    42  
    43  
    44     #-------------------------------------------------------------------- 
    45  
    46     class SimpleTest < EventMachine::Protocols::HeaderAndContentProtocol 
    47         attr_reader :first_header, :my_headers, :request 
    48  
    49         def receive_first_header_line hdr 
    50             @first_header ||= [] 
    51             @first_header << hdr 
    52         end 
    53         def receive_headers hdrs 
    54             @my_headers ||= [] 
    55             @my_headers << hdrs 
    56         end 
    57         def receive_request hdrs, content 
    58             @request ||= [] 
    59             @request << [hdrs, content] 
    60         end 
    61     end 
    62  
    63     def test_no_content 
    64         Thread.abort_on_exception = true 
    65         the_connection = nil 
    66         EventMachine.run { 
    67             EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
    68                 the_connection = conn 
    69             end 
    70             EventMachine.add_timer(4) {raise "test timed out"} 
    71             EventMachine.defer proc { 
    72                 t = TCPSocket.new TestHost, TestPort 
    73                 t.write [ 
    74                     "aaa\n", "bbb\r\n", "ccc\n", "\n" 
    75                 ].join 
    76                 t.close 
    77                 if RUBY_VERSION =~ /\A1\.9\./ 
    78                         sleep 0.1 
    79                         STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
     43        TestHost = "127.0.0.1" 
     44        TestPort = 8905 
     45 
     46 
     47        #-------------------------------------------------------------------- 
     48 
     49        class SimpleTest < EventMachine::Protocols::HeaderAndContentProtocol 
     50                attr_reader :first_header, :my_headers, :request 
     51 
     52                def receive_first_header_line hdr 
     53                        @first_header ||= [] 
     54                        @first_header << hdr 
    8055                end 
    81             }, proc { 
    82                 EventMachine.stop 
    83             } 
    84         } 
    85         assert_equal( ["aaa"], the_connection.first_header ) 
    86         assert_equal( [%w(aaa bbb ccc)], the_connection.my_headers ) 
    87         assert_equal( [[%w(aaa bbb ccc), ""]], the_connection.request ) 
    88     end 
    89  
    90     def test_content 
    91         Thread.abort_on_exception = true 
    92         the_connection = nil 
    93         content = "A" * 50 
    94         headers = ["aaa", "bbb", "Content-length: #{content.length}", "ccc"] 
    95         EventMachine.run { 
    96             EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
    97                 the_connection = conn 
    98             end 
    99             EventMachine.add_timer(4) {raise "test timed out"} 
    100             EventMachine.defer proc { 
    101                 t = TCPSocket.new TestHost, TestPort 
    102                 headers.each {|h| t.write "#{h}\r\n" } 
    103                 t.write "\n" 
    104                 t.write content 
    105                 t.close 
    106                 if RUBY_VERSION =~ /\A1\.9\./ 
    107                         sleep 0.1 
    108                         STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
     56                def receive_headers hdrs 
     57                        @my_headers ||= [] 
     58                        @my_headers << hdrs 
    10959                end 
    110             }, proc { 
    111                 EM.stop 
    112             } 
    113         } 
    114         assert_equal( ["aaa"], the_connection.first_header ) 
    115         assert_equal( [headers], the_connection.my_headers ) 
    116         assert_equal( [[headers, content]], the_connection.request ) 
    117     end 
    118  
    119     def test_several_requests 
    120         Thread.abort_on_exception = true 
    121         the_connection = nil 
    122         content = "A" * 50 
    123         headers = ["aaa", "bbb", "Content-length: #{content.length}", "ccc"] 
    124         EventMachine.run { 
    125             EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
    126                 the_connection = conn 
    127             end 
    128             EventMachine.add_timer(4) {raise "test timed out"} 
    129             EventMachine.defer proc { 
    130                 t = TCPSocket.new TestHost, TestPort 
    131                 5.times { 
    132                     headers.each {|h| t.write "#{h}\r\n" } 
    133                     t.write "\n" 
    134                     t.write content 
    135                 } 
    136                 t.close 
    137                 if RUBY_VERSION =~ /\A1\.9\./ 
    138                         sleep 0.1 
    139                         STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
     60                def receive_request hdrs, content 
     61                        @request ||= [] 
     62                        @request << [hdrs, content] 
    14063                end 
    141             }, proc { 
    142                 EventMachine.stop 
    143             } 
    144         } 
    145         assert_equal( ["aaa"] * 5, the_connection.first_header ) 
    146         assert_equal( [headers] * 5, the_connection.my_headers ) 
    147         assert_equal( [[headers, content]] * 5, the_connection.request ) 
    148     end 
     64        end 
     65 
     66 
     67        def test_no_content 
     68                Thread.abort_on_exception = true 
     69                the_connection = nil 
     70                EventMachine.run { 
     71                        EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
     72                                the_connection = conn 
     73                        end 
     74                        EventMachine.add_timer(4) {raise "test timed out"} 
     75 
     76                        pr = proc { 
     77                                t = TCPSocket.new TestHost, TestPort 
     78                                t.write [ "aaa\n", "bbb\r\n", "ccc\n", "\n" ].join 
     79                                t.close 
     80                        } 
     81 
     82                        if RUBY_PLATFORM =~ /java/i 
     83                                pr.call 
     84                                EM.add_timer(0.5) {EM.stop} 
     85                        else 
     86                                EventMachine.defer proc { 
     87                                        pr.call 
     88                                        if RUBY_VERSION =~ /\A1\.9\./ 
     89                                                sleep 0.1 
     90                                                STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
     91                                        end 
     92                                }, proc { 
     93                                        EventMachine.stop 
     94                                } 
     95                        end 
     96                } 
     97                assert_equal( ["aaa"], the_connection.first_header ) 
     98                assert_equal( [%w(aaa bbb ccc)], the_connection.my_headers ) 
     99                assert_equal( [[%w(aaa bbb ccc), ""]], the_connection.request ) 
     100        end 
     101 
     102 
     103 
     104 
     105        def test_content 
     106                Thread.abort_on_exception = true 
     107                the_connection = nil 
     108                content = "A" * 50 
     109                headers = ["aaa", "bbb", "Content-length: #{content.length}", "ccc"] 
     110                EventMachine.run { 
     111                        EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
     112                                the_connection = conn 
     113                        end 
     114                        EventMachine.add_timer(4) {raise "test timed out"} 
     115 
     116                        pr = proc { 
     117                                t = TCPSocket.new TestHost, TestPort 
     118                                headers.each {|h| t.write "#{h}\r\n" } 
     119                                t.write "\n" 
     120                                t.write content 
     121                                t.close 
     122                        } 
     123 
     124                        if RUBY_PLATFORM =~ /java/i 
     125                                # I/O on threads seems completely unreliable in Java. 
     126                                pr.call 
     127                                EM.add_timer(0.5) {EM.stop} 
     128                        else 
     129                                EventMachine.defer proc { 
     130                                        pr.call 
     131                                        if RUBY_VERSION =~ /\A1\.9\./ 
     132                                                sleep 0.1 
     133                                                STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
     134                                        end 
     135                                }, proc { 
     136                                        EM.stop 
     137                                } 
     138                        end 
     139                } 
     140                assert_equal( ["aaa"], the_connection.first_header ) 
     141                assert_equal( [headers], the_connection.my_headers ) 
     142                assert_equal( [[headers, content]], the_connection.request ) 
     143        end 
     144 
     145 
     146 
     147        def test_several_requests 
     148                Thread.abort_on_exception = true 
     149                the_connection = nil 
     150                content = "A" * 50 
     151                headers = ["aaa", "bbb", "Content-length: #{content.length}", "ccc"] 
     152                EventMachine.run { 
     153                        EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
     154                                the_connection = conn 
     155                        end 
     156                        EventMachine.add_timer(4) {raise "test timed out"} 
     157 
     158                        pr = proc { 
     159                                t = TCPSocket.new TestHost, TestPort 
     160                                5.times { 
     161                                        headers.each {|h| t.write "#{h}\r\n" } 
     162                                        t.write "\n" 
     163                                        t.write content 
     164                                } 
     165                                t.close 
     166                        } 
     167 
     168                        if RUBY_PLATFORM =~ /java/i 
     169                                pr.call 
     170                                EM.add_timer(1) {EM.stop} 
     171                        else 
     172                                EventMachine.defer proc { 
     173                                        pr.call 
     174                                        if RUBY_VERSION =~ /\A1\.9\./ 
     175                                                sleep 0.1 
     176                                                STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
     177                                        end 
     178                                }, proc { 
     179                                        EventMachine.stop 
     180                                } 
     181                        end 
     182                } 
     183                assert_equal( ["aaa"] * 5, the_connection.first_header ) 
     184                assert_equal( [headers] * 5, the_connection.my_headers ) 
     185                assert_equal( [[headers, content]] * 5, the_connection.request ) 
     186        end 
    149187 
    150188 
    …… 
    172210    end 
    173211 
    174     def test_interpret_headers 
    175         Thread.abort_on_exception = true 
    176         the_connection = nil 
    177         content = "A" * 50 
    178         headers = [ 
    179             "GET / HTTP/1.0", 
    180             "Accept: aaa", 
    181             "User-Agent: bbb", 
    182             "Host:          ccc", 
    183             "x-tempest-header:ddd" 
    184         ] 
    185         EventMachine.run { 
    186             EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
    187                 the_connection = conn 
    188             end 
    189             EventMachine.add_timer(4) {raise "test timed out"} 
    190             EventMachine.defer proc { 
    191                 t = TCPSocket.new TestHost, TestPort 
    192                 headers.each {|h| t.write "#{h}\r\n" } 
    193                 t.write "\n" 
    194                 t.write content 
    195                 t.close 
    196                 if RUBY_VERSION =~ /\A1\.9\./ 
    197                         sleep 0.1 
    198                         STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
    199                 end 
    200             }, proc { 
    201                 EventMachine.stop 
    202             } 
    203         } 
    204  
    205         hsh = the_connection.headers_2_hash( the_connection.my_headers.shift ) 
    206         assert_equal( 
    207             { 
    208                 :accept => "aaa", 
    209                 :user_agent => "bbb", 
    210                 :host => "ccc", 
    211                 :x_tempest_header => "ddd" 
    212             }, 
    213             hsh 
    214          ) 
    215     end 
    216  
    217     #-------------------------------------------------------------------- 
     212        def test_interpret_headers 
     213                Thread.abort_on_exception = true 
     214                the_connection = nil 
     215                content = "A" * 50 
     216                headers = [ 
     217                        "GET / HTTP/1.0", 
     218                        "Accept: aaa", 
     219                        "User-Agent: bbb", 
     220                        "Host:      ccc", 
     221                        "x-tempest-header:ddd" 
     222                ] 
     223 
     224                EventMachine.run { 
     225                        EventMachine.start_server( TestHost, TestPort, SimpleTest ) do |conn| 
     226                                the_connection = conn 
     227                        end 
     228                        EventMachine.add_timer(4) {raise "test timed out"} 
     229 
     230                        pr = proc { 
     231                                t = TCPSocket.new TestHost, TestPort 
     232                                headers.each {|h| t.write "#{h}\r\n" } 
     233                                t.write "\n" 
     234                                t.write content 
     235                                t.close 
     236                        } 
     237 
     238                        if RUBY_PLATFORM =~ /java/i 
     239                                pr.call 
     240                                EM.add_timer(0.5) {EM.stop} 
     241                        else 
     242                                EventMachine.defer proc { 
     243                                        pr.call 
     244                                        if RUBY_VERSION =~ /\A1\.9\./ 
     245                                                sleep 0.1 
     246                                                STDERR.puts "Introducing extraneous sleep for Ruby 1.9" 
     247                                        end 
     248                                }, proc { 
     249                                        EventMachine.stop 
     250                                } 
     251                        end 
     252                } 
     253 
     254                hsh = the_connection.headers_2_hash( the_connection.my_headers.shift ) 
     255                assert_equal( 
     256                        { 
     257                                :accept => "aaa", 
     258                                :user_agent => "bbb", 
     259                                :host => "ccc", 
     260                                :x_tempest_header => "ddd" 
     261                        }, 
     262                        hsh 
     263                ) 
     264        end 
     265 
    218266 
    219267end 

  • branches/raggi/tests/test_sasl.rb

    r712 r723  
    5454                resp = nil 
    5555                EM.run { 
    56                         EM::Timer.new(2) {EM.stop} 
    5756                        EM.start_server( Host, Port, SaslServer ) 
    5857 
    5958                        c = EM.connect( Host, Port, SaslClient ) 
    6059                        d = c.validate?( TestUser, TestPsw ) 
     60                        d.timeout 2 
    6161                        d.callback { 
    6262                                resp = true 

  • branches/raggi/tests/test_timers.rb

    r668 r723  
    108108 
    109109        # This test is only applicable to compiled versions of the reactor. 
    110         # Pure ruby versions have no built-in limit on the number of outstanding timers. 
     110        # Pure ruby and java versions have no built-in limit on the number of outstanding timers. 
    111111        # 
    112112        def test_timer_change_max_outstanding 
    …… 
    118118                EM.run { 
    119119                        if EM.library_type == :pure_ruby 
     120                                ten_thousand_timers.call 
     121                        elsif EM.library_type == :java 
    120122                                ten_thousand_timers.call 
    121123                        else