Class: Cuboid::RPC::Server::Agent::Node

Inherits:
Object
  • Object
show all
Includes:
UI::Output
Defined in:
lib/cuboid/rpc/server/agent/node.rb

Overview

Agent node class, helps maintain a list of all available Agents in the grid and announce itself to peering Agents.

As soon as a new Node is fired up it checks-in with its peer and grabs a list of all available peers.

As soon as it receives the peer list it then announces itself to them.

Upon convergence there will be a grid of Agents each one with its own copy of all available Agent URLs.

Author:

Instance Method Summary collapse

Methods included from UI::Output

#error_buffer, initialize, #log_error, #output_provider_file, #print_bad, #print_debug, #print_error, #print_info, #print_line, #print_ok, #print_status, #print_verbose, #reroute_to_file, #reroute_to_file?

Methods included from UI::OutputInterface

initialize

Methods included from UI::OutputInterface::Personalization

#included

Methods included from UI::OutputInterface::Controls

#debug?, #debug_level, #debug_level_1?, #debug_level_2?, #debug_level_3?, #debug_level_4?, #debug_off, #debug_on, initialize, #verbose?, #verbose_off, #verbose_on

Methods included from UI::OutputInterface::ErrorLogging

#error_logfile, #has_error_log?, initialize, #set_error_logfile

Methods included from UI::OutputInterface::Implemented

#print_debug_backtrace, #print_debug_exception, #print_debug_level_1, #print_debug_level_2, #print_debug_level_3, #print_debug_level_4, #print_error_backtrace, #print_exception

Methods included from UI::OutputInterface::Abstract

#output_provider_file, #print_bad, #print_debug, #print_error, #print_info, #print_line, #print_ok, #print_status, #print_verbose

Constructor Details

#initialize(options, server, logfile = nil) ⇒ Node

Initializes the node by:

* Adding the peer (if the user has supplied one) to the peer list.
* Getting the peer's peer list and appending them to its own.
* Announces itself to the peer and instructs it to propagate our URL
  to the others.

Parameters:

  • options (Cuboid::Options)
  • server (Server::Base) —

    Agent’s RPC server.

  • logfile (String) (defaults to: nil) —

    Where to send the output.



34
35
36
37
38
39
40
41
42
43
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
71
72
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 34

def initialize( options, server, logfile = nil )
    @options = options
    @server  = server
    @url     = @server.url

    reroute_to_file( logfile ) if logfile

    print_status 'Initializing grid node...'

    @dead_nodes = Set.new
    @peers = Set.new
    @nodes_info_cache = []

    if (peer = @options.agent.peer)
        # Grab the peer's peers.
        connect_to_peer( peer ).peers do |urls|
            if urls.rpc_exception?
                add_dead_peer( peer )
                print_info "Neighbour seems dead: #{peer}"
                add_dead_peer( peer )
                next
            end

            # Add peer and announce it to everyone.
            add_peer( peer, true )

            urls.each { |url| @peers << url if url != @url }
        end
    end

    print_status 'Node ready.'

    log_updated_peers

    Raktr.global.at_interval( @options.agent.ping_interval ) do
        ping
        check_for_comebacks
    end
end

Instance Method Details

#add_peer(node_url, propagate = false) ⇒ Object

Adds a peer to the peer list.

Parameters:

  • node_url (String) —

    URL of a peering node.

  • propagate (Boolean) (defaults to: false) —

    Whether or not to announce the new node to the peers.



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 97

def add_peer( node_url, propagate = false )
    # we don't want ourselves in the Set
    return false if node_url == @url
    return false if @peers.include?( node_url )

    print_status "Adding peer: #{node_url}"

    @peers << node_url
    log_updated_peers
    announce( node_url ) if propagate

    connect_to_peer( node_url ).add_peer( @url, propagate ) do |res|
        next if !res.rpc_exception?
        add_dead_peer( node_url )
        print_status "Neighbour seems dead: #{node_url}"
    end
    true
end

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


175
176
177
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 175

def alive?
    true
end

#grid_member? ⇒ Boolean

Returns ‘true` if grid member, `false` otherwise.

Returns:

  • (Boolean) —

    ‘true` if grid member, `false` otherwise.



76
77
78
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 76

def grid_member?
    @peers.any?
end

#info ⇒ Hash

Returns * ‘url` – This node’s URL.

  • ‘name` – Nickname

  • ‘peers` – Array of peers.

Returns:

  • (Hash) —

    • ‘url` – This node’s URL.

    • ‘name` – Nickname

    • ‘peers` – Array of peers.



166
167
168
169
170
171
172
173
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 166

def info
    {
        'url'                    => @url,
        'name'                   => @options.agent.name,
        'peers'             => @peers.to_a,
        'unreachable_peers' => @dead_nodes.to_a
    }
end

#peers ⇒ Array

Returns Neighbour/node/peer URLs.

Returns:

  • (Array) —

    Neighbour/node/peer URLs.



124
125
126
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 124

def peers
    @peers.to_a
end

#peers_with_info(&block) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 128

def peers_with_info( &block )
    fail 'This method requires a block!' if !block_given?

    @peers_cmp = ''

    if @nodes_info_cache.empty? || @peers_cmp != peers.to_s
        @peers_cmp = peers.to_s

        each = proc do |peer, iter|
            connect_to_peer( peer ).info do |info|
                if info.rpc_exception?
                    print_info "Neighbour seems dead: #{peer}"
                    add_dead_peer( peer )
                    log_updated_peers

                    iter.return( nil )
                else
                    iter.return( info )
                end
            end
        end

        after = proc do |nodes|
            @nodes_info_cache = nodes.compact
            block.call( @nodes_info_cache )
        end

        Raktr.global.create_iterator( peers ).map( each, after )
    else
        block.call( @nodes_info_cache )
    end
end

#remove_peer(url) ⇒ Object 



116
117
118
119
120
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 116

def remove_peer( url )
    @peers.delete url
    @dead_nodes.delete url
    nil
end

#unplug ⇒ Object 



80
81
82
83
84
85
86
87
88
89
 
# File 'lib/cuboid/rpc/server/agent/node.rb', line 80

def unplug
    @peers.each do |peer|
        connect_to_peer( peer ).remove_peer( @url ) {}
    end

    @peers.clear
    @dead_nodes.clear

    nil
end