Package core :: Module streams

[hide private]
[frames] | no frames]

# Source Code for Module core.streams

      1  # Copyright 2011 the original author or authors.
      2  #
      3  # Licensed under the Apache License, Version 2.0 (the "License");
      4  # you may not use this file except in compliance with the License.
      5  # You may obtain a copy of the License at
      6  #
      7  #      http://www.apache.org/licenses/LICENSE-2.0
      8  #
      9  # Unless required by applicable law or agreed to in writing, software
     10  # distributed under the License is distributed on an "AS IS" BASIS,
     11  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     12  # See the License for the specific language governing permissions and
     13  # limitations under the License.
     14
     15  import org.vertx.java.core.streams.Pump
     16
     17  from core.handlers import BufferHandler, StreamEndHandler
     18
     19  __author__ = "Scott Horn"
     20  __email__ = "scott@hornmicro.com"
     21  __credits__ = "Based entirely on work by Tim Fox http://tfox.org"
     22
     23 -class WriteStream(object):
     24      """
     25       A mixin module which represents a stream of data that can be written to.
     26
     27       Any class that mixes in this module can be used by a  to pump data from a  to it.
     28
     29      """
     30
     31 -    def write_buffer(self, buff):
     32          """Write some data to the stream. The data is put on an internal write queue, and the write actually happens
     33          asynchronously. To avoid running out of memory by putting too much on the write queue,
     34          check the  method before writing. This is done automatically if using a .
     35          param [Buffer]. The buffer to write.
     36          """
     37          self.java_obj.writeBuffer(buff._to_java_buffer())
     38
     39 -    def set_write_queue_max_size(self, size):
     40          """Set the maximum size of the write queue. You will still be able to write to the stream even
     41          if there is more data than this in the write queue. This is used as an indicator by classes such as
     42          to provide flow control.
     43
     44          Keyword arguments:
     45          @param size: The maximum size, in bytes.
     46          """
     47          self.java_obj.setWriteQueueMaxSize(size)
     48
     49      write_queue_max_size = property(fset=set_write_queue_max_size)
     50
     51      @property
     52 -    def write_queue_full(self):
     53          """Is the write queue full?
     54
     55          return True if there are more bytes in the write queue than the max write queue size.
     56          """
     57          return self.java_obj.writeQueueFull()
     58
     59 -    def drain_handler(self, handler):
     60          """Set a drain handler on the stream. If the write queue is full, then the handler will be called when the write
     61          queue has been reduced to maxSize / 2. See  for an example of this being used.
     62
     63          Keyword arguments:
     64          @param handler: The drain handler
     65          """
     66          self.java_obj.drainHandler(BufferHandler(handler))
     67
     68 -    def exception_handler(self, handler):
     69          """Set an execption handler on the stream.
     70
     71          Keyword arguments:
     72          @param handler: The exception handler
     73          """
     74          self.java_obj.exceptionHandler(ExceptionHandler(handler))
     75
     76 -    def _to_write_stream(self):
     77          return self.java_obj
     78
     79 -class ReadStream(object):
     80      """A mixin module which represents a stream of data that can be read from.
     81
     82      Any class that mixes in this module can be used by a  to pump data from a  to it.
     83      """
     84
     85 -    def data_handler(self, handler):
     86          """Set a data handler. As data is read, the handler will be called with the data.
     87
     88          Keyword arguments:
     89          @param handler: The data handler
     90          """
     91          self.java_obj.dataHandler(BufferHandler(handler))
     92
     93 -    def pause(self):
     94          """Pause the ReadStream. After calling this, the ReadStream will aim to send no more data to the """
     95          self.java_obj.pause()
     96
     97 -    def resume(self):
     98          """Resume reading. If the ReadStream has been paused, reading will recommence on it."""
     99          self.java_obj.resume()
    100
    101 -    def exception_handler(self, handler):
    102          """Set an execption handler on the stream.
    103          param [Block] hndlr. The exception handler
    104          """
    105          self.java_obj.exceptionHandler(ExceptionHandler(handler))
    106
    107 -    def end_handler(self, handler):
    108          """Set an end handler on the stream. Once the stream has ended, and there is no more data to be read, this handler will be called.
    109
    110          Keyword arguments:
    111          @param handler: The exception handler"""
    112          self.java_obj.endHandler(StreamEndHandler(handler))
    113
    114 -    def _to_read_stream(self):
    115          return self.java_obj
    116
    117 -class Pump(object):
    118      """Pumps data from a ReadStream to a WriteStream and performs flow control where necessary to
    119      prevent the write stream from getting overloaded.
    120
    121      Instances of this class read bytes from a ReadStream and write them to a WriteStream. If data
    122      can be read faster than it can be written this could result in the write queue of the WriteStream growing
    123      without bound, eventually causing it to exhaust all available RAM.
    124      To prevent this, after each write, instances of this class check whether the write queue of the WriteStream
    125      is full, and if so, the ReadStream is paused, and a WriteStreamdrain_handler is set on the WriteStream.
    126      When the WriteStream has processed half of its backlog, the drain_handler will be called,
    127      which results in the pump resuming the ReadStream.
    128
    129      This class can be used to pump from any ReadStream to any  WriteStream,
    130      e.g. from an HttpServerRequest to an AsyncFile, or from NetSocket to a WebSocket.
    131      """
    132
    133 -    def __init__(self, read_stream, write_stream):
    134          #raise "read_stream is not a ReadStream" if !read_stream.is_a? ReadStream
    135          #raise "write_stream is not a WriteStream" if !write_stream.is_a? WriteStream
    136          self.j_rs = read_stream._to_read_stream()
    137          self.j_ws = write_stream._to_write_stream()
    138          self.j_pump = org.vertx.java.core.streams.Pump.createPump(self.j_rs, self.j_ws)
    139
    140 -    def set_write_queue_max_size(self, val):
    141          """Set the write queue max size
    142
    143          Keyword arguments:
    144          @param val: The write queue max size
    145          """
    146          self.j_pump.setWriteQueueMaxSize(val)
    147
    148      write_queue_max_size = property(fset=set_write_queue_max_size)
    149
    150 -    def start(self):
    151          """Start the Pump. The Pump can be started and stopped multiple times."""
    152          self.j_pump.start()
    153
    154 -    def stop(self):
    155          """Stop the Pump. The Pump can be started and stopped multiple times."""
    156          self.j_pump.stop()
    157
    158      @property
    159 -    def bytes_pumped(self):
    160          """return the total number of bytes pumped by this pump."""
    161          return self.j_pump.getBytesPumped()
    162