:sockets:: socket list
 :clean:: clean UTF8 strings or provide block to run on every read string
 :pool:: ActionPool to use
 Creates a new watcher for sockets
 start the watcher
112
def start
      if(@sockets.size < 0)
        raise 'No sockets available for listening'
      elsif(!@runner.nil? && @runner.alive?)
        raise AlreadyRunning.new
      else
        @stop = false
        @runner = Thread.new{watch}
      end
    end
1166108
async def _watch(self, node, conn, names):
        ""
        for name in names:
            slot = self._determine_slot('WATCH', name)
            dist_node = self.connection_pool.get_node_by_slot(slot)
            if node.get('name') != dist_node['name']:
                # raise error if commands in a transaction can not hash to same node
                if len(node) > 0:
                    raise ClusterTransactionError("Keys in request don't hash to the same node")
        if self.explicit_transaction:
            raise RedisError('Cannot issue a WATCH after a MULTI')
        await conn.send_command('WATCH', *names)
        return await conn.read_response()
390753
def run
      while true
        message = @socket.read_multipart
        if @socket.is_a? Celluloid::ZMQ::RouterSocket
          message = message[1]
        else
          message = message[0]
        end
        handle_message message
      end
    end
334668
def lex
      r = []
      while (tok = @lexer.lex).kind != T::EOF
        r.push(tok)
      end
      r
    end
967306
def consume_string(ending = nil)
      ending = @s.current if ending.nil?
      value  = String.new

      until @s.eos?
        case char = @s.consume
        when ending
          break

        when "\n"
          # Parse error.
          @s.reconsume
          return create_token(:bad_string,
            :error => true,
            :value => value)

        when '\\'
          case @s.peek
          when ''
            # End of the input, so do nothing.
            next

          when "\n"
            @s.consume

          else
            value << consume_escaped
          end

        else
          value << char
        end
      end

      create_token(:string, :value => value)
    end
816825
def clean!
      HotTub.logger.info "[HotTub] Cleaning pool #{@name}!" if HotTub.logger
      @mutex.synchronize do
        begin
          @_pool.each do |clnt|
            clean_client(clnt)
          end
        ensure
          @cond.signal
        end
      end
      nil
    end
622807
public synchronized void checkIn(SocketBox sb) {

        if (((ManagedSocketBox) sb).getStatus() != ManagedSocketBox.BUSY) {
            throw new IllegalArgumentException("The socket      is already marked free, cannot check it in twice.");
        }

        if (!busySockets.containsKey(sb)) {
            throw new IllegalArgumentException("This socket does not exist in the pool of busy sockets.");
        }

        if (freeSockets.containsKey(sb)) {
            throw new IllegalArgumentException("This socket already exists in the pool of free sockets.");
        }

        if (! ((ManagedSocketBox)sb).isReusable()) {

            throw new IllegalArgumentException("This socket is not reusable; cannot check in.");

        }

        ((ManagedSocketBox) sb).setStatus(ManagedSocketBox.FREE);
        busySockets.remove(sb);
        freeSockets.put(sb, sb);

    }
1008053
def each_raw_message_line offset
    @mutex.synchronize do
      ensure_open
      @f.seek offset
      until @f.eof? || MBox::is_break_line?(l = @f.gets)
        yield l
      end
    end
  end
580335
def process
      read_events.each do |event|
        event.callback!
        event.flags.include?(:ignored) && event.notifier.watchers.delete(event.watcher_id)
      end
    end
659122
def remove_stale_sockets(self):
        """"""
        if self.opts.max_idle_time_seconds is not None:
            with self.lock:
                while (self.sockets and
                       self.sockets[-1].idle_time_seconds() > self.opts.max_idle_time_seconds):
                    sock_info = self.sockets.pop()
                    sock_info.close()
        while True:
            with self.lock:
                if (len(self.sockets) + self.active_sockets >=
                        self.opts.min_pool_size):
                    # There are enough sockets in the pool.
                    break

            # We must acquire the semaphore to respect max_pool_size.
            if not self._socket_semaphore.acquire(False):
                break
            try:
                sock_info = self.connect()
                with self.lock:
                    self.sockets.appendleft(sock_info)
            finally:
                self._socket_semaphore.release()
620184
def start
      write_header(@socket, build_header)
      read_header(@socket)
      @thread = Thread.start do
        while @is_running
          data = read_all(@socket)
          msg = @topic_type.new
          msg.deserialize(data)
          @byte_received += data.length
          @msg_queue.push(msg)
        end
      end
    end
1175127
public void init(Socket s)
  {
    _s = s;
    
    _is = null;
    _os = null;
    _needsFlush = false;
  }
1095147
def read_header(socket)
      header = ::ROS::TCPROS::Header.new
      header.deserialize(read_all(socket))
      header
    end
645955
def go_stdlib(self):
    
    out = self._go_dist.create_go_cmd('list', args=['std']).check_output()
    return frozenset(out.decode('utf-8').strip().split())
310961
def receive(max_len=nil)
      max_len ||= 1024

      begin
        data = @socket.recv(max_len)
      rescue IO::WaitReadable
        @logger.logv 'Retrying receive...'
        IO.select([@socket])
        retry
      rescue Exception => e
        if e.class != 'IOError' && e.message != 'closed stream'
          @logger.log_exception(e, 'EXCEPTION during receive!')
        end
      end

      if (data.nil? || data.empty?)
        @logger.log "Client #{@socket.inspect} disconnected!"
        data = ''
      else
        @logger.log "Received #{data.length} bytes"
      end

      return data
    end
955391
def socket_ready(hostport):
    
    if hasattr(hostport, 'startswith'):
        # string representation of the hostport combination
        if hostport.startswith('tcp://'):
            hostport = hostport[6:]  # strip tcp://
        host, port = hostport.split(':')
        hostport = (host, int(port))
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        exc = sock.connect_ex(hostport)
    finally:
        sock.close()
    return False if exc else True
817327
def <<(stream)
      case stream_type(stream)
      when :client then
        return unless stream.connected?
        jid = stream.user.jid.bare
        @clients[jid] ||= []
        @clients[jid] << stream
      when :server then @servers << stream
      when :component then @components << stream
      end
    end