Class: Thrift::TInboundStream

Inherits:
TBaseStream show all
Defined in:
lib/thrift/inbound_stream.rb

Instance Method Summary collapse

Methods inherited from TBaseStream

#ready

Constructor Details

#initialize(iprot, oprot, name, seqid, klass, message_type, unlock = nil) ⇒ TInboundStream

Returns a new instance of TInboundStream.



3
4
5
6
7
8
9
 
# File 'lib/thrift/inbound_stream.rb', line 3

def initialize(iprot, oprot, name, seqid, klass, message_type, unlock = nil)
  @message_type = message_type
  @klass = klass
  @unlock = unlock
  @closing = false
  super(iprot, oprot, name, seqid)
end

Instance Method Details

#close ⇒ Object 



11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/thrift/inbound_stream.rb', line 11

def close
  wait_ready

  @mutex.synchronize do
    return nil if @closed

    begin
      @closing = true
      write_shell_unlocked @message_type + 1

      loop do
        mtype = read_message_begin

        case mtype
        when @message_type
          @iprot.skip(Types::STRUCT)
          @iprot.read_message_end
        when @message_type + 2
          @iprot.read_message_end
          return
        else
          raise ApplicationException.new(
            ApplicationException::INVALID_MESSAGE_TYPE,
            'invalid message type'
          )
        end
      end
    ensure
      @unlock&.call
    end
  end
end

#receive ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/thrift/inbound_stream.rb', line 44

def receive
  wait_ready

  @mutex.synchronize do
    case read_message_begin
    when @message_type
      res = @klass.new
      res.read(@iprot)
      @iprot.read_message_end

      return res.arg
    when @message_type + 1
      @iprot.read_message_end
      write_shell_unlocked @message_type + 2 unless @closing
      @closed = true
      @cond.broadcast
      @unlock&.call

      raise EOFError
    else
      raise ApplicationException.new(
        ApplicationException::INVALID_MESSAGE_TYPE,
        'invalid message type'
      )
    end
  end
end