Class: Thrift::TBidiStream

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

Instance Method Summary collapse

Constructor Details

#initialize(iprot, oprot, name, seqid, inbound_klass, outbound_klass, inbound_message_type, outbound_message_type, unlock = nil) ⇒ TBidiStream

Returns a new instance of TBidiStream.



31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/thrift/bidi_stream.rb', line 31

def initialize(iprot, oprot, name, seqid, inbound_klass, outbound_klass, inbound_message_type, outbound_message_type, unlock = nil)
  @inbound_message_type = inbound_message_type
  @outbound_message_type = outbound_message_type
  @inbound_klass = inbound_klass
  @outbound_klass = outbound_klass
  @inbound_closed = false
  @outbound_closed = false
  @message_queue = Queue.new
  @unlock = unlock
  @closing_inbound = false
  @closing_outbound = false
  super(iprot, oprot, name, seqid)
end

Instance Method Details

#close ⇒ Object 



107
108
109
 
# File 'lib/thrift/bidi_stream.rb', line 107

def close
  mark_as_closed
end

#inbound_close ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/thrift/bidi_stream.rb', line 50

def inbound_close
  wait_ready

  @mutex.synchronize do
    return if @closed || @inbbound_closed

    @closing_inbound = true
    write_shell_unlocked @inbound_message_type + 1

    loop do
      return if @closed || @inbound_closed

      @cond.wait(@mutex)
    end
  end
end

#outbound_close ⇒ Object 



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/thrift/bidi_stream.rb', line 67

def outbound_close
  wait_ready

  @mutex.synchronize do
    return if @closed || @outbound_closed

    @closing_outbound = true
    write_shell_unlocked @outbound_message_type + 1

    loop do
      return nil if @closed || @outbound_closed

      @cond.wait(@mutex)
    end
  end
end

#ready ⇒ Object 



45
46
47
48
 
# File 'lib/thrift/bidi_stream.rb', line 45

def ready
  Thread.new { background_read }
  super
end

#receive ⇒ Object

Raises:

  • (EOFError) 


84
85
86
87
88
89
90
91
92
 
# File 'lib/thrift/bidi_stream.rb', line 84

def receive
  raise EOFError if @closed || @inbound_closed

  msg = @message_queue.pop

  raise EOFError unless msg

  msg.arg
end

#send(msg) ⇒ Object

Raises:

  • (EOFError) 


94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/thrift/bidi_stream.rb', line 94

def send(msg)
  raise EOFError if @closed || @outbound_closed

  unless @oprot.trans.open?
    mark_as_closed
    raise EOFError
  end

  wait_ready

  write @outbound_message_type, @outbound_klass.new(arg: msg)
end