Class: Takagi::Observer::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/takagi/observer/client.rb,
sig/takagi/observer/client.rbs

Overview

Lightweight CoAP observe client for integration testing.

Instance Method Summary collapse

Constructor Details

#initialize(uri) ⇒ Client

Returns a new instance of Client.

Parameters:

  • uri (Object) 


11
12
13
14
15
16
 
# File 'lib/takagi/observer/client.rb', line 11

def initialize(uri)
  @uri = URI.parse(uri)
  @token = SecureRandom.hex(4)
  @socket = UDPSocket.new
  @on_notify = nil
end

Instance Method Details

#handle_notification_iteration ⇒ Object

Returns:

  • (Object) 


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

def handle_notification_iteration
  data, _addr = @socket.recvfrom_nonblock(1024)
  inbound = Takagi::Message::Inbound.new(data)
  return unless inbound.token == @token

  payload = inbound.payload
  payload = payload.split("\xFF", 2).last if payload&.include?("\xFF")
  payload = payload&.dup&.force_encoding('UTF-8')
  Takagi.logger.info "Received notify: #{payload}"
  @on_notify&.call(payload, inbound)
rescue IO::WaitReadable
  @socket.wait_readable
  retry
rescue IOError
  @running = false
end

#listen_for_notifications ⇒ Object

Returns:

  • (Object) 


45
46
47
48
 
# File 'lib/takagi/observer/client.rb', line 45

def listen_for_notifications
  @running = true
  @thread = Thread.new { handle_notification_iteration while @running }
end

#on_notify {|arg0| ... } ⇒ Object

Yields:

Yield Parameters:

  • arg0

Yield Returns:

  • (Object)

Returns:

  • (Object) 


18
19
20
 
# File 'lib/takagi/observer/client.rb', line 18

def on_notify(&block)
  @on_notify = block
end

#send_observe_request ⇒ Object

Returns:

  • (Object) 


35
36
37
38
39
40
41
42
43
 
# File 'lib/takagi/observer/client.rb', line 35

def send_observe_request
  message = Takagi::Message::Request.new(
    method: :get,
    uri: @uri,
    token: @token,
    observe: 0
  )
  @socket.send(message.to_bytes, 0, @uri.host, @uri.port || 5683)
end

#stop ⇒ Object Also known as: close

Returns:

  • (Object) 


27
28
29
30
 
# File 'lib/takagi/observer/client.rb', line 27

def stop
  @running = false
  @socket.close unless @socket.closed?
end

#subscribe ⇒ Object

Returns:

  • (Object) 


22
23
24
25
 
# File 'lib/takagi/observer/client.rb', line 22

def subscribe
  send_observe_request
  listen_for_notifications
end