класс Rinda::RingServer

Родитель:: Объект
Включенные модули:: DRb::DRbUndumped

«RingServer» позволяет определить местоположение «Rinda::TupleSpace» посредством UDP-рассылок. По умолчанию расположение сервиса происходит по следующему алгоритму:

«RingServer» начинает прослушивать сеть по UDP-адресу широковещательной рассылки.
«RingFinger» отправляет UDP-пакет, содержащий DRb URI, по которому будет ожидаться ответ.
«RingServer» получает UDP-пакет и подключается обратно к указанному DRb URI с помощью сервиса DRb.

«RingServer» требует TupleSpace:

ts = Rinda::TupleSpace.new
rs = Rinda::RingServer.new

«RingServer» также может прослушивать объявления по многоадресным адресам. Это позволяет запускать несколько RingServer на одном узле. Для использования широковещательной и многоадресной рассылки по сети:

ts = Rinda::TupleSpace.new
rs = Rinda::RingServer.new ts, %w[Socket::INADDR_ANY, 239.0.0.1 ff02::1]

Общедоступные методы класса

new(ts, addresses=[Socket::INADDR_ANY], port=Ring_PORT) Показать исходный код

# File lib/rinda/ring.rb, line 94
def initialize(ts, addresses=[Socket::INADDR_ANY], port=Ring_PORT)
  @port = port

  if Integer === addresses then
    addresses, @port = [Socket::INADDR_ANY], addresses
  end

  @renewer = Renewer.new

  @ts = ts
  @sockets = []
  addresses.each do |address|
    if Array === address
      make_socket(*address)
    else
      make_socket(address)
    end
  end

  @w_services = write_services
  @r_service  = reply_service
end

Объявляет ts по указанным addresses по адресу port.

Если addresses опущено, используется только адрес UDP-широковещательной рассылки.

addresses может содержать несколько адресов. Если в addresses указан многоадресной адрес, то «RingServer» будет прослушивать запросы по многоадресной рассылке.

Если вы используете IPv4 многоадресную рассылку, вам может потребоваться установить адрес входного интерфейса, который присоединяется к группе многоадресной рассылки.

ts = Rinda::TupleSpace.new
rs = Rinda::RingServer.new(ts, [['239.0.0.1', '9.5.1.1']])

Вы можете установить адреса как Array Object. Первый элемент Array — это многоадресной адрес, а второй — адрес входного интерфейса. Если второй опушен, используется ‘0.0.0.0’.

Если вы используете IPv6 многоадресную рассылку, вам может потребоваться установить адрес локального интерфейса и индекс входного интерфейса:

rs = Rinda::RingServer.new(ts, [['ff02::1', '::1', 1]])

Первый элемент — многоадресной адрес, второй — адрес входного интерфейса, третий — индекс входного интерфейса.

В настоящее время нет простого способа получить индекс интерфейса по имени.

Если второй опушен, используется ‘::1’. Если третий опушен, используется 0 (интерфейс по умолчанию).

Общедоступные методы экземпляра

do_reply() Показать исходный код

# File lib/rinda/ring.rb, line 218
def do_reply
  tuple = @ts.take([:lookup_ring, nil], @renewer)
  Thread.new { tuple[1].call(@ts) rescue nil}
rescue
end

Извлекает кортежи поиска из TupleSpace и отправляет их DRb объекта адрес локального TupleSpace.

do_write(msg) Показать исходный код

# File lib/rinda/ring.rb, line 193
def do_write(msg)
  Thread.new do
    begin
      tuple, sec = Marshal.load(msg)
      @ts.write(tuple, sec)
    rescue
    end
  end
end

Извлекает ответ URI из msg и добавляет его в TupleSpace, где он будет получен reply_service для уведомления.

make_socket(address, interface_address=nil, multicast_interface=0) Показать исходный код

# File lib/rinda/ring.rb, line 132
def make_socket(address, interface_address=nil, multicast_interface=0)
  addrinfo = Addrinfo.udp(address, @port)

  socket = Socket.new(addrinfo.pfamily, addrinfo.socktype,
                      addrinfo.protocol)

  if addrinfo.ipv4_multicast? or addrinfo.ipv6_multicast? then
    if Socket.const_defined?(:SO_REUSEPORT) then
      socket.setsockopt(:SOCKET, :SO_REUSEPORT, true)
    else
      socket.setsockopt(:SOCKET, :SO_REUSEADDR, true)
    end

    if addrinfo.ipv4_multicast? then
      interface_address = '0.0.0.0' if interface_address.nil?
      socket.bind(Addrinfo.udp(interface_address, @port))

      mreq = IPAddr.new(addrinfo.ip_address).hton +
        IPAddr.new(interface_address).hton

      socket.setsockopt(:IPPROTO_IP, :IP_ADD_MEMBERSHIP, mreq)
    else
      interface_address = '::1' if interface_address.nil?
      socket.bind(Addrinfo.udp(interface_address, @port))

      mreq = IPAddr.new(addrinfo.ip_address).hton +
        [multicast_interface].pack('I')

      socket.setsockopt(:IPPROTO_IPV6, :IPV6_JOIN_GROUP, mreq)
    end
  else
    socket.bind(addrinfo)
  end

  socket
rescue
  socket = socket.close if socket
  raise
ensure
  @sockets << socket if socket
end

Создает сокет по адресу address

Если address — многоадресной адрес, то interface_address и multicast_interface могут быть установлены как необязательные.

Созданный сокет привязывается к interface_address. Если вы используете IPv4 многоадресную рассылку, то интерфейс interface_address используется как входной интерфейс. Если interface_address опущено или равно null, то используется ‘0.0.0.0’ или ‘::1’.

Если вы используете IPv6 многоадресную рассылку, то multicast_interface используется как входной интерфейс. multicast_interface — индекс сетевого интерфейса. Если multicast_interface опущено, используется 0 (интерфейс по умолчанию).

reply_service() Показать исходный код

# File lib/rinda/ring.rb, line 206
def reply_service
  Thread.new do
    loop do
      do_reply
    end
  end
end

Создаёт поток, который уведомляет ожидающих клиентов из TupleSpace.

shutdown() Показать исходный код

# File lib/rinda/ring.rb, line 227
def shutdown
  @renewer.renew = false

  @w_services.each do |thread|
    thread.kill
    thread.join
  end

  @sockets.each do |socket|
    socket.close
  end

  @r_service.kill
  @r_service.join
end

Выключает «RingServer»

write_services() Показать исходный код

# File lib/rinda/ring.rb, line 178
def write_services
  @sockets.map do |s|
    Thread.new(s) do |socket|
      loop do
        msg = socket.recv(1024)
        do_write(msg)
      end
    end
  end
end

Создаёт потоки, которые собирают UDP-пакеты и передают их в do_write для декодирования.

Ruby Core © 1993–2022 Yukihiro Matsumoto
Licensed under the Ruby License.
Ruby Standard Library © contributors
Licensed under their own licenses.