import org.jgroups.View; //导入依赖的package包/类
public void addListener(@NotNull final JgroupsViewListener viewListener) {
    requireNonNull(viewListener);
    log.debug("Adding view listener {}...", viewListener);

    synchronized (viewListeners) {
        if (viewListeners.containsKey(viewListener)) {
            throw new IllegalStateException("View listener is already registered.");
        }
    }

    View initialView = jChannel.view();
    viewListener.initialView(initialView.getMembers());

    synchronized (viewListeners) {
        viewListeners.put(viewListener, initialView);
    }
}
 

import org.jgroups.View; //导入依赖的package包/类
public void viewUpdate(final View newView) {
    requireNonNull(newView);
    log.debug("Processing view update {}...", newView);

    Map<JgroupsViewListener, Address[][]> diffs = new HashMap<>();

    synchronized (viewListeners) {
        viewListeners.forEach((viewListener, currentView) -> {
            Address[][] diff = View.diff(currentView, newView);
            viewListeners.put(viewListener, newView);
            diffs.put(viewListener, diff);
        });
    }

    diffs.forEach((viewListener, diff) -> {
        for (Address joined : diff[0]) {
            viewListener.memberJoined(joined);
        }
        for (Address left : diff[1]) {
            viewListener.memberLeft(left);
        }
    });
}
 

import org.jgroups.View; //导入依赖的package包/类
protected void handleView(View new_view, View old_view, boolean coord_changed) {
    if(is_coord) {
        if(clear_table_on_view_change)
            clearTable();
        else if(old_view != null && new_view != null) {
            Address[][] diff=View.diff(old_view, new_view);
            Address[] left_mbrs=diff[1];
            for(Address left_mbr : left_mbrs)
                if(left_mbr != null && !new_view.containsMember(left_mbr))
                    remove(left_mbr);
        }
    }
    if(coord_changed || clear_table_on_view_change)
        writeOwnInformation(); // write immediately
    if(info_writer_max_writes_after_view > 0)
        startInfoWriter(); // and / or write in the background
}
 

import org.jgroups.View; //导入依赖的package包/类
@Override
public void installView(NetView v) {
  this.view = v;

  if (this.jgAddress.getVmViewId() < 0) {
    this.jgAddress.setVmViewId(this.localAddress.getVmViewId());
  }
  List<JGAddress> mbrs = new ArrayList<>(v.size());
  mbrs.addAll(v.getMembers().stream().map(JGAddress::new).collect(Collectors.toList()));
  ViewId vid = new ViewId(new JGAddress(v.getCoordinator()), v.getViewId());
  View jgv = new View(vid, new ArrayList<>(mbrs));
  logger.trace("installing JGroups view: {}", jgv);
  this.myChannel.down(new Event(Event.VIEW_CHANGE, jgv));

  addressesWithIoExceptionsProcessed.clear();
  if (encrypt != null) {
    encrypt.installView(v);
  }
}
 

import org.jgroups.View; //导入依赖的package包/类
@Override
public void viewAccepted(View view) {
    if (endpoint.isEnableViewMessages()) {
        Exchange exchange = endpoint.createExchange(view);
        try {
            LOG.debug("Processing view: {}", view);
            processor.process(exchange, new AsyncCallback() {
                @Override
                public void done(boolean doneSync) {
                    // noop
                }
            });
        } catch (Exception e) {
            throw new JGroupsException("Error in consumer while dispatching exchange containing view " + view, e);
        }
    } else {
        LOG.debug("Option enableViewMessages is set to false. Skipping processing of the view: {}", view);
    }
}
 

import org.jgroups.View; //导入依赖的package包/类
/**
 * Creates predicate rejecting messages that are instances of {@code org.jgroups.View}, but have not been received
 * by the coordinator JGroups node. This filter is useful for keeping only view messages indicating that receiving
 * endpoint is a master node.
 *
 * @return predicate filtering out non-coordinator view messages.
 */
public static Predicate dropNonCoordinatorViews() {
    return new Predicate() {
        @Override
        public boolean matches(Exchange exchange) {
            Object body = exchange.getIn().getBody();
            LOG.debug("Filtering message {}.", body);
            if (body instanceof View) {
                View view = (View) body;
                Address coordinatorNodeAddress =  view.getMembers().get(COORDINATOR_NODE_INDEX);
                Address channelAddress = exchange.getIn().getHeader(HEADER_JGROUPS_CHANNEL_ADDRESS, Address.class);
                LOG.debug("Comparing endpoint channel address {} against the coordinator node address {}.",
                        channelAddress, coordinatorNodeAddress);
                return channelAddress.equals(coordinatorNodeAddress);
            }
            LOG.debug("Body {} is not an instance of org.jgroups.View . Skipping filter.", body);
            return false;
        }
    };
}
 

import org.jgroups.View; //导入依赖的package包/类
public void start() throws Exception {
    String props="udp.xml";

    channel=new JChannel(props);

    channel.setReceiver(new ReceiverAdapter() {
        public void viewAccepted(View view) {
            setInternalState(view.getMembers());
        }

        public void setInternalState(java.util.List<Address> mbrs) {
            members.clear();
            for(Address mbr : mbrs)
                addNode(mbr);
            coordinator=mbrs.size() <= 1 || (mbrs.size() > 1 && mbrs.iterator().next().equals(my_addr));
            repaint();
        }
    });
    channel.connect(channel_name);
    my_addr=channel.getAddress();
    if(my_addr != null)
        setTitle(my_addr.toString());
    pack();
    setVisible(true);
}
 

import org.jgroups.View; //导入依赖的package包/类
protected void handleView(View new_view, View old_view, boolean coord_changed) {
    if(is_coord) {
        if(clear_table_on_view_change)
            clearTable();
        else if(old_view != null && new_view != null) {
            Address[][] diff=View.diff(old_view, new_view);
            Address[] left_mbrs=diff[1];
            for(Address left_mbr : left_mbrs)
                if(left_mbr != null && !new_view.containsMember(left_mbr))
                    remove(cluster_name, left_mbr);
        }
    }
    if(coord_changed || clear_table_on_view_change)
        writeOwnInformation(true); // write immediately
    if(info_writer_max_writes_after_view > 0)
        startInfoWriter(); // and / or write in the background
}
 

import org.jgroups.View; //导入依赖的package包/类
public void readFrom(DataInput in) throws Exception {
    byte flags=in.readByte();

    // 1. view
    if((flags & VIEW_PRESENT) == VIEW_PRESENT) {
        view=new View();
        view.readFrom(in);
    }

    // 2. digest
    if((flags & DIGEST_PRESENT) == DIGEST_PRESENT) {
        digest=new Digest(view.getMembersRaw());
        digest.readFrom(in, false);
    }

    // 3. fail_reason
    if((flags & FAIL_REASON_PRESENT) == FAIL_REASON_PRESENT)
        fail_reason=in.readUTF();
}
 

import org.jgroups.View; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public Object down(Event evt) {
    switch(evt.getType()) {
        case Event.MSG:
            Message msg=(Message)evt.getArg();
            if(msg.isFlagSet(Message.Flag.NO_FC))
                break;
            int length=msg.getLength();
            if(length == 0)
                break;
            return handleDownMessage(evt, msg, length);
        case Event.CONFIG:
            handleConfigEvent((Map<String,Object>)evt.getArg()); 
            break;
        case Event.VIEW_CHANGE:
            handleViewChange(((View)evt.getArg()).getMembers());
            break;
    }
    return down_prot.down(evt); // this could potentially use the lower protocol's thread which may block
}
 

import org.jgroups.View; //导入依赖的package包/类
public Object down(Event evt) {
    switch(evt.getType()) {
        case Event.SET_LOCAL_ADDRESS:
            local_addr=(Address)evt.getArg();
            break;
        case Event.VIEW_CHANGE:
            handleView((View)evt.getArg());
            break;
    }

    if(num_event_handlers > 0) {
        for(EventHandler handler: event_handlers) {
            try {
                handler.down(evt);
            }
            catch(Throwable t) {
                log.error("event handler failed handling down event", t);
            }
        }
    }

    return down_prot.down(evt);
}
 

import org.jgroups.View; //导入依赖的package包/类
public Object up(Event evt) {
    switch(evt.getType()) {
        case Event.VIEW_CHANGE:
            handleView((View)evt.getArg());
            break;
    }
    if(num_event_handlers > 0) {
        for(EventHandler handler: event_handlers) {
            try {
                handler.up(evt);
            }
            catch(Throwable t) {
                log.error("event handler failed handling up event", t);
            }
        }
    }
    return up_prot.up(evt);
}
 

import org.jgroups.View; //导入依赖的package包/类
/**
 * Fragment a packet if larger than frag_size (add a header). Otherwise just pass down. Only
 * add a header if framentation is needed !
 */
public Object down(Event evt) {
    switch(evt.getType()) {

        case Event.MSG:
            Message msg=(Message)evt.getArg();
            long size=msg.size();
            num_sent_msgs++;
            if(size > frag_size) {
                if(log.isTraceEnabled()) {
                    StringBuilder sb=new StringBuilder("message size is ");
                    sb.append(size).append(", will fragment (frag_size=").append(frag_size).append(')');
                    log.trace(sb.toString());
                }
                fragment(msg, size);  // Fragment and pass down
                return null;
            }
            break;

        case Event.VIEW_CHANGE:
            handleViewChange((View)evt.getArg());
            break;
    }

    return down_prot.down(evt);  // Pass on to the layer below us
}
 

import org.jgroups.View; //导入依赖的package包/类
/**
 * If event is a message, if it is fragmented, re-assemble fragments into big message and pass up the stack.
 */
public Object up(Event evt) {
    switch(evt.getType()) {
        case Event.MSG:
            Message msg=(Message)evt.getArg();
            FragHeader hdr=(FragHeader)msg.getHeader(this.id);
            if(hdr != null) { // needs to be defragmented
                Message assembled_msg=unfragment(msg, hdr);
                if(assembled_msg != null)
                    up_prot.up(new Event(Event.MSG, assembled_msg));
                return null;
            }
            else {
                num_received_msgs++;
            }
            break;

        case Event.VIEW_CHANGE:
            handleViewChange((View)evt.getArg());
            break;
    }

    return up_prot.up(evt); // Pass up to the layer above us by default
}
 

import org.jgroups.View; //导入依赖的package包/类
/**
 * Fragment a packet if larger than frag_size (add a header). Otherwise just pass down. Only
 * add a header if fragmentation is needed !
 */
public Object down(Event evt) {
    switch(evt.getType()) {

        case Event.MSG:
            Message msg=(Message)evt.getArg();
            long size=msg.getLength();
            if(size > frag_size) {
                fragment(msg);  // Fragment and pass down
                return null;
            }
            break;

        case Event.VIEW_CHANGE:
            handleViewChange((View)evt.getArg());
            break;

        case Event.SET_LOCAL_ADDRESS:
            local_addr=(Address)evt.getArg();
            break;
    }

    return down_prot.down(evt);  // Pass on to the layer below us
}
 

import org.jgroups.View; //导入依赖的package包/类
/**
 * If event is a message, if it is fragmented, re-assemble fragments into big message and pass up the stack.
 */
public Object up(Event evt) {
    switch(evt.getType()) {

        case Event.MSG:
            Message msg=(Message)evt.getArg();
            FragHeader hdr=(FragHeader)msg.getHeader(this.id);
            if(hdr != null) { // needs to be defragmented
                Message assembled_msg=unfragment(msg, hdr);
                if(assembled_msg != null) {
                    if(log.isTraceEnabled()) log.trace("%s: assembled_msg is %s", local_addr, assembled_msg);
                    assembled_msg.setSrc(msg.getSrc()); // needed ? YES, because fragments have a null src !!
                    up_prot.up(new Event(Event.MSG, assembled_msg));
                }
                return null;
            }
            break;

        case Event.VIEW_CHANGE:
            handleViewChange((View)evt.getArg());
            break;
    }

    return up_prot.up(evt); // Pass up to the layer above us by default
}
 

import org.jgroups.View; //导入依赖的package包/类
protected void handleView(View new_view, View old_view, boolean coord_changed) {
    if(is_coord) {
        if(coord_changed) {
            if(remove_all_files_on_view_change)
                removeAll(cluster_name);
            else if(remove_old_coords_on_view_change) {
                Address old_coord=old_view != null? old_view.getCreator() : null;
                if(old_coord != null)
                    remove(cluster_name, old_coord);
            }
        }
        if(coord_changed || View.diff(old_view, new_view)[1].length > 0) {
            writeAll();
            if(remove_all_files_on_view_change || remove_old_coords_on_view_change)
                startInfoWriter();
        }
    }
    else if(coord_changed) // I'm no longer the coordinator
        remove(cluster_name, local_addr);
}
 

import org.jgroups.View; //导入依赖的package包/类
public void viewAccepted(View new_view) {
    nodes.clear();
    for(Address node: new_view.getMembers()) {
        int hash=Math.abs(node.hashCode() & (HASH_SPACE - 1));
        for(int i=hash; i < hash + HASH_SPACE; i++) {
            short new_index=(short)(i & (HASH_SPACE - 1));
            if(!nodes.containsKey(new_index)) {
                nodes.put(new_index, node);
                break;
            }
        }
    }

    if(log.isTraceEnabled()) {
        StringBuilder sb=new StringBuilder("node mappings:\n");
        for(Map.Entry<Short,Address> entry: nodes.entrySet()) {
            sb.append(entry.getKey() + ": " + entry.getValue()).append("\n");
        }
        log.trace(sb);
    }
}
 

import org.jgroups.View; //导入依赖的package包/类
public void viewAccepted(final View new_view) {
    final List<Address> old_nodes=this.view != null? new ArrayList<>(this.view.getMembers()) : null;

    this.view=new_view;
    if(log.isDebugEnabled())
        log.debug("new view: " + new_view);

    if(hash_function != null)
        hash_function.installNodes(new_view.getMembers());

    for(MembershipListener l: membership_listeners)
        l.viewAccepted(new_view);

    if(old_nodes != null) {
        timer.schedule(new Runnable() {
            public void run() {
                rebalance(old_nodes, new ArrayList<>(new_view.getMembers()));
            }
        }, 100, TimeUnit.MILLISECONDS);
    }
}
 

import org.jgroups.View; //导入依赖的package包/类
public void testNonConflictingMerge() {
    Address ip1=Util.createRandomAddress("x"), ip2=Util.createRandomAddress("y");
    View tmp_view=View.create(a1,1,a1,a2,a3,ip1,ip2);
    MutableDigest cons_d=new MutableDigest(tmp_view.getMembersRaw());

    cons_d.set(ip1,10,10);
    cons_d.set(ip2,20,20);
    cons_d.merge(d);

    Assert.assertEquals(cons_d.capacity(), 5);

    Assert.assertEquals(cons_d.get(ip1)[0], 10);
    Assert.assertEquals(cons_d.get(ip2)[0], 20);
    Assert.assertEquals(cons_d.get(a1)[0], 500);
    Assert.assertEquals(cons_d.get(a2)[0], 26);
    Assert.assertEquals(cons_d.get(a3)[0], 25);

    Assert.assertEquals(cons_d.get(ip1)[1], 10);
    Assert.assertEquals(cons_d.get(ip2)[1], 20);
    Assert.assertEquals(cons_d.get(a1)[1], 501);
    Assert.assertEquals(cons_d.get(a2)[1], 26);
    Assert.assertEquals(cons_d.get(a3)[1], 33);
}