Class: Nl::NotificationRouter

Inherits:
Object
  • Object
show all
Defined in:
nl/lib/nl/notification_router.rb

Overview

Routes unsolicited frames to per-family notification channels.

Instance Method Summary collapse

Constructor Details

#initialize(protocol:, capacity:) ⇒ NotificationRouter

Returns a new instance of NotificationRouter.



10
11
12
13
14
15
16
17
 
# File 'nl/lib/nl/notification_router.rb', line 10

def initialize(protocol:, capacity:)
  @protocol = protocol
  @capacity = capacity
  @mutex = Mutex.new
  @entries = {}
  @routes = {}
  @closed = false
end

Instance Method Details

#channel(endpoint) ⇒ Object 



41
42
43
44
45
 
# File 'nl/lib/nl/notification_router.rb', line 41

def channel(endpoint)
  @mutex.synchronize do
    @entries.fetch(@protocol.notification_channel_key(endpoint)).channel
  end
end

#close ⇒ Object 



72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'nl/lib/nl/notification_router.rb', line 72

def close
  entries = @mutex.synchronize do
    return nil if @closed

    @closed = true
    old = @entries.values
    @entries.clear
    @routes.clear
    old
  end
  entries.each { it.channel.close }
  nil
end

#lose_all(error) ⇒ Object 



86
87
88
89
90
 
# File 'nl/lib/nl/notification_router.rb', line 86

def lose_all(error)
  entries = @mutex.synchronize { @entries.values.dup }
  entries.each { it.channel.fail(error) }
  nil
end

#register(endpoint, classes) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'nl/lib/nl/notification_router.rb', line 19

def register(endpoint, classes)
  @mutex.synchronize do
    raise ClosedError, 'notification router is closed' if @closed

    channel_key = @protocol.notification_channel_key(endpoint)
    entry = @entries[channel_key]
    merged_classes = entry ? entry.classes.merge(classes) : classes.dup
    route_keys = @protocol.notification_route_keys(endpoint, merged_classes)
    if route_key = route_keys.find { @routes[it] && !@routes[it].equal?(entry) }
      raise ArgumentError, "notification route #{route_key.inspect} is already registered"
    end

    unless entry
      entry = Entry.new(endpoint, merged_classes, NotificationChannel.new(capacity: @capacity))
      @entries[channel_key] = entry
    end
    entry.classes.replace(merged_classes)
    route_keys.each { @routes[it] = entry }
    entry.channel
  end
end

#route(header, payload) ⇒ Object

Returns true if the frame belongs to a registered notification family.



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

def route(header, payload)
  if header.type == Raw::NLMSG_OVERRUN
    entries = @mutex.synchronize { @entries.values.dup }
    entries.each { it.channel.fail(NotificationLossError.new('kernel reported Netlink overrun')) }
    return true
  end

  entry = @mutex.synchronize do
    @routes[@protocol.notification_frame_key(header)]
  end
  return false unless entry
  return false unless @protocol.notification_frame?(entry.endpoint, header, payload)

  message_class = @protocol.notification_class(entry.endpoint, header, payload, entry.classes)
  return true unless message_class

  notification = @protocol.decode_notification(entry.endpoint, header, payload, message_class)
  entry.channel.push(notification)
  true
rescue => error
  entry&.channel&.fail(error)
  true
end