For live replication. Reads live from the local hyperlog and continues to send new nodes to the other end.
553636
function () {
    var write = function (node, enc, cb) {
      node.value = encoder.encode(node.value, dag.valueEncoding)
      stream.node(node, cb)
    }

    stream.emit('live')
    pipe(dag.createReadStream({since: changes, live: true}), through.obj(write))
  }
947262
private void sendLocalRestoreInformation(Long max, Set<SnapshotInfo> snapshots) {
        String jsonData = serializeRestoreInformation(max, snapshots);
        String zkNode = VoltZK.restore + "/" + m_hostId;
        try {
            m_zk.create(zkNode, jsonData.getBytes(StandardCharsets.UTF_8),
                        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        } catch (Exception e) {
            throw new RuntimeException("Failed to create Zookeeper node: " +
                                       e.getMessage(), e);
        }
    }
855462
def set(self, uri, content, **meta):
        
        try:
            node = self._update(uri, content, **meta)
            created = False
        except NodeDoesNotExist:
            node = self._create(uri, content, **meta)
            created = True
        return self._serialize(uri, node), created
131013
def apply(self, node):
        
        new_node = self.run(node)
        return self.update, new_node
337176
def atlas_get_live_neighbors( remote_peer_hostport, peer_table=None, min_health=MIN_PEER_HEALTH, min_request_count=1 ):
    

    with AtlasPeerTableLocked(peer_table) as ptbl:
        alive_peers = []
        for peer_hostport in ptbl.keys():
            if peer_hostport == remote_peer_hostport:
                continue

            num_reqs = atlas_peer_get_request_count( peer_hostport, peer_table=ptbl )
            if num_reqs < min_request_count:
                continue

            health = atlas_peer_get_health( peer_hostport, peer_table=ptbl )
            if health < min_health:
                continue

            alive_peers.append( peer_hostport )

    random.shuffle(alive_peers)
    return alive_peers
1310861
def nodes_changed?(new_nodes)
      return true if address_for(@master) != new_nodes[:master]
      return true if different?(addresses_for(@slaves), new_nodes[:slaves])
      false
    end
551578
def _handle_continue(self, node, scope, ctxt, stream):
        
        self._dlog("handling continue")
        raise errors.InterpContinue()
1298130
def readlines(self):
        
        continuation = False
        while True:
            yield self.readline(continuation)
            continuation = True
126983
def add_nodes(self, nodes):
        '''
        '''
        new_nodes, master_map = self._add_nodes_as_master(nodes)

        for n in new_nodes:
            if n.name not in master_map:
                continue
            master_name = master_map[n.name]
            target = self.get_node(master_name)
            if not target:
                raise NodeNotFound(master_name)
            n.replicate(target.name)
            n.flush_cache()
            target.flush_cache()
1293103
long recordRead(int bufferIndex, Node<K, V> node) {
        // The location in the buffer is chosen in a racy fashion as the increment
        // is not atomic with the insertion. This means that concurrent reads can
        // overlap and overwrite one another, resulting in a lossy buffer.
        final AtomicLong counter = readBufferWriteCount[bufferIndex];
        final long writeCount = counter.get();
        counter.lazySet(writeCount + 1);

        final int index = (int) (writeCount & READ_BUFFER_INDEX_MASK);
        readBuffers[bufferIndex][index].lazySet(node);

        return writeCount;
    }
1201367
def nodes
      nodes = []
      cache 'nodes' do
        connection.send_data("nodes")
        while ( ( line = connection.read_line ) != "." )
          nodes << line
        end
        nodes
      end
    end
1204867
def _wait_nodes_updated(self, cluster_member, new_nodes):
        '''
        '''
        while True:
            known_nodes = [n['name'] for n in cluster_member.nodes()]
            added_nodes = [n.name for n in new_nodes]
            if len(set(known_nodes) & set(added_nodes)) == len(added_nodes):
                break;
            logger.info('waiting for adding new nodes in `cluster nodes`')
            time.sleep(1)
1067744
private long grow(PrintStream out, List<ItemSet> list, TotalSupportTree ttree, HeaderTableItem header, int[] itemset, int[] localItemSupport, int[] prefixItemset) {
        long n = 1;
        int support = header.count;
        int item = header.id;
        itemset = insert(itemset, item);

        collect(out, list, ttree, itemset, support);
        
        if (header.node.next == null) {
            FPTree.Node node = header.node;
            n += grow(out, list, ttree, node.parent, itemset, support);
        } else {
            // Count singles in linked list
            if (getLocalItemSupport(header.node, localItemSupport)) {
                // Create local FP tree
                FPTree fptree = getLocalFPTree(header.node, localItemSupport, prefixItemset);
                // Mine new FP-tree
                n += grow(out, list, ttree, fptree, itemset, localItemSupport, prefixItemset);
            }
        }

        return n;
    }
1189611
def update_nodes_published(self):
        
        if self.pk:
            self.node_set.all().update(is_published=self.is_published)
1146735
void afterRead(Node<K, V> node) {
        final int bufferIndex = readBufferIndex();
        final long writeCount = recordRead(bufferIndex, node);
        drainOnReadIfNeeded(bufferIndex, writeCount);
        notifyListener();
    }
315520
def updateNodeCapabilities(self, nodeId, node, vendorSpecific=None):
        
        response = self.updateNodeCapabilitiesResponse(nodeId, node, vendorSpecific)
        return self._read_boolean_response(response)
218560
def replicate_to(other_db, continuous = false, create_target = false, doc_ids = nil)
      replicate(other_db, continuous, :source => name, :create_target => create_target, :doc_ids => doc_ids)
    end