import org.jgroups.Event; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public Object up(Event evt) {

  switch (evt.getType()) {

    case Event.FIND_MBRS:
      List<Address> missing = (List<Address>) evt.getArg();

      Responses responses = new Responses(false);
      for (Address laddr : missing) {
        try {
          if (laddr instanceof JGAddress) {
            PingData pd = new PingData(laddr, true, laddr.toString(), newIpAddress(laddr));
            responses.addResponse(pd, false);
            updateUDPCache(pd);
          }
        } catch (RuntimeException e) {
          logger.warn("Unable to create PingData response", e);
          throw e;
        }
      }
      return responses;
  }
  return up_prot.up(evt);
}
 

import org.jgroups.Event; //导入依赖的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.Event; //导入依赖的package包/类
/** look for certain messages that may need to be altered before being sent */
void filterOutgoingMessage(DistributionMessage m) {
  switch (m.getDSFID()) {
    case JOIN_RESPONSE:
      JoinResponseMessage jrsp = (JoinResponseMessage) m;

      if (jrsp.getRejectionMessage() == null
          && services.getConfig().getTransport().isMcastEnabled()) {
        // get the multicast message digest and pass it with the join response
        Digest digest = (Digest) this.myChannel.getProtocolStack().getTopProtocol()
            .down(Event.GET_DIGEST_EVT);
        HeapDataOutputStream hdos = new HeapDataOutputStream(500, Version.CURRENT);
        try {
          digest.writeTo(hdos);
        } catch (Exception e) {
          logger.fatal("Unable to serialize JGroups messaging digest", e);
        }
        jrsp.setMessengerData(hdos.toByteArray());
      }
      break;
    default:
      break;
  }
}
 

import org.jgroups.Event; //导入依赖的package包/类
void filterIncomingMessage(DistributionMessage m) {
  switch (m.getDSFID()) {
    case JOIN_RESPONSE:
      JoinResponseMessage jrsp = (JoinResponseMessage) m;

      if (jrsp.getRejectionMessage() == null
          && services.getConfig().getTransport().isMcastEnabled()) {
        byte[] serializedDigest = jrsp.getMessengerData();
        ByteArrayInputStream bis = new ByteArrayInputStream(serializedDigest);
        DataInputStream dis = new DataInputStream(bis);
        try {
          Digest digest = new Digest();
          digest.readFrom(dis);
          logger.trace("installing JGroups message digest {}", digest);
          this.myChannel.getProtocolStack().getTopProtocol()
              .down(new Event(Event.MERGE_DIGEST, digest));
          jrsp.setMessengerData(null);
        } catch (Exception e) {
          logger.fatal("Unable to read JGroups messaging digest", e);
        }
      }
      break;
    default:
      break;
  }
}
 

import org.jgroups.Event; //导入依赖的package包/类
public void initialize() {
  receivedAcks = new ConcurrentHashSet<>();

  pingPonger = new GMSPingPonger();
  // UUID logicalAddress = (UUID) channel.getAddress();
  // IpAddress ipaddr = (IpAddress) channel.down(new Event(Event.GET_PHYSICAL_ADDRESS));
  //
  // myAddress = new JGAddress(logicalAddress, ipaddr);
  myAddress = (JGAddress) channel.down(new Event(Event.GET_LOCAL_ADDRESS));

  addressConversionMap = new ConcurrentHashMap<>(this.lastView.size());
  List<InternalDistributedMember> members = this.lastView.getMembers();
  for (InternalDistributedMember addr : members) {
    SocketAddress sockaddr =
        new InetSocketAddress(addr.getNetMember().getInetAddress(), addr.getPort());
    addressConversionMap.put(sockaddr, addr);
  }

  isDebugEnabled = logger.isDebugEnabled();
  resume();
}
 

import org.jgroups.Event; //导入依赖的package包/类
private void handleMessage(Message msg) {
  if (collectMessages) {
    collectedMessages.add(msg);
  }
  Object o = msg.getHeader(nakackHeaderId);
  if (o != null) {
    mcastSentDataMessages++;
  } else {
    o = msg.getHeader(unicastHeaderId);
    if (o != null) {
      UNICAST3.Header hdr = (UNICAST3.Header) o;
      switch (hdr.type()) {
        case UNICAST3.Header.DATA:
          unicastSentDataMessages++;
          Message response = new Message(uuid, msg.getDest(), null);
          response.putHeader(unicastHeaderId, UNICAST3.Header.createAckHeader(hdr.seqno(),
              hdr.connId(), System.currentTimeMillis()));
          up_prot.up(new Event(Event.MSG, response));
          break;
      }
    }
  }
}
 

import org.jgroups.Event; //导入依赖的package包/类
@Test
public void recorderHandlesRejectedExecution() throws Exception {
  Message msg = mock(Message.class);
  when(msg.getHeader(any(Short.class))).thenReturn(Header.createDataHeader(1L, (short) 1, true));
  when(msg.size()).thenReturn(150L);

  // GEODE-1178, the TP protocol may throw a RejectedExecutionException & StatRecorder should
  // retry
  when(mockDownProtocol.down(any(Event.class))).thenThrow(new RejectedExecutionException());

  // after the first down() throws an exception we want StatRecorder to retry, so
  // we set the Manager to say no shutdown is in progress the first time and then say
  // one IS in progress so we can break out of the StatRecorder exception handling loop
  when(services.getCancelCriterion()).thenReturn(new Services().getCancelCriterion());
  Manager manager = mock(Manager.class);
  when(services.getManager()).thenReturn(manager);
  when(manager.shutdownInProgress()).thenReturn(Boolean.FALSE, Boolean.TRUE);

  verify(mockDownProtocol, never()).down(isA(Event.class));

  Event evt = new Event(Event.MSG, msg);
  recorder.down(evt);

  verify(mockDownProtocol, times(2)).down(isA(Event.class));
}
 

import org.jgroups.Event; //导入依赖的package包/类
@Override
public Object up(Event event) {

    int type = event.getType();

    if (displayUpMessages) {

        if (type == Event.MSG) {

            Message msg = (Message) event.getArg();

            if (msg != null) {

                byte[] buffer = msg.getBuffer();

                if (buffer != null) {
                    log.info(">>>" + new String(buffer));
                }
            }
        }
    }
    return up_prot.up(event);
}
 

import org.jgroups.Event; //导入依赖的package包/类
public Object down(Event evt) {
    switch(evt.getType()) {
        case Event.MSG:
            Message msg=(Message)evt.getArg();
            FORK.ForkHeader hdr=(FORK.ForkHeader)msg.getHeader(FORK.ID);
            if(hdr == null)
                msg.putHeader(FORK.ID, hdr=new FORK.ForkHeader(fork_stack_id, null));
            else
                hdr.setForkStackId(fork_stack_id);
            break;
        case Event.SET_LOCAL_ADDRESS:
        case Event.VIEW_CHANGE:
        case Event.CONNECT:
        case Event.CONNECT_USE_FLUSH:
        case Event.CONNECT_WITH_STATE_TRANSFER:
        case Event.CONNECT_WITH_STATE_TRANSFER_USE_FLUSH:
        case Event.DISCONNECT:
            return null; // don't propagate further down, this is only important for the main stack
    }
    return down_prot.down(evt);
}
 

import org.jgroups.Event; //导入依赖的package包/类
public void sendMessages(final Protocol prot, final int start, final int end) {
    final Thread sender=new Thread() {
        public void run() {
            for(int i=start; i <= end; i++) {
                Message msg=new Message(null, i);
                System.out.println("[" + prot.getValue("local_addr") + "] --> sending message " + i);
                prot.down(new Event(Event.MSG,msg));
            }
        }
    };
    sender.setName("BytemanSenderThread");
    sender.start();
    try {
        sender.join(1000);
    }
    catch(InterruptedException e) {
    }
}
 

import org.jgroups.Event; //导入依赖的package包/类
/**
   * An event was received from the layer below. Usually the current layer will want to examine
   * the event type and - depending on its type - perform some computation
   * (e.g. removing headers from a MSG event type, or updating the internal membership list
   * when receiving a VIEW_CHANGE event).
   * Finally the event is either a) discarded, or b) an event is sent down
   * the stack using <code>down_prot.down()</code> or c) the event (or another event) is sent up
   * the stack using <code>up_prot.up()</code>.
   */
  public Object up(Event evt) {
switch(evt.getType()) {
          case Event.MSG:
		Message message = (Message)evt.getArg();
		if ( message.isFlagSet( Message.Flag.OOB ) ) {
			return up_prot.up(evt);
		}
		else {
			PrioHeader hdr=(PrioHeader)message.getHeader(id);
			if(hdr != null) {
				log.trace("%s: adding priority message %d to UP queue", local_addr, hdr.getPriority());
				upMessageQueue.add( new PriorityMessage( evt, hdr.getPriority() ) );
				// send with hdr.prio
				return null;
			}
      			return up_prot.up(evt);
              }
          default:
      		return up_prot.up(evt);
      }
  }
 

import org.jgroups.Event; //导入依赖的package包/类
@Test
public void testResponse() throws Exception {
    Address local_addr = pinger.getLocalAddress();
    PhysicalAddress physical_addr = (PhysicalAddress) pinger
            .down(new Event(Event.GET_PHYSICAL_ADDRESS, local_addr));
    PingData data = createPingData(local_addr, physical_addr);
    final PingHeader hdr = getPingHeader(data);
    Message msg = new Message(null).setFlag(Message.Flag.DONT_BUNDLE)
            .putHeader(pinger.getId(), hdr).setBuffer(streamableToBuffer(data));
    URL url = new URL("http://localhost:8888");
    HttpURLConnection conn = (HttpURLConnection) url.openConnection(Proxy.NO_PROXY);
    conn.addRequestProperty(Server.CLUSTER_NAME, TestBase.CLUSTER_NAME);
    conn.setDoOutput(true);
    conn.setRequestMethod("POST");

    DataOutputStream out = new DataOutputStream(conn.getOutputStream());
    msg.writeTo(out);
    out.flush();

    Assert.assertEquals(200, conn.getResponseCode());
}
 

import org.jgroups.Event; //导入依赖的package包/类
/**
 * Sends up a multiple messages in a {@link MessageBatch}. The sender of the batch is always the same, and so is the
 * destination (null == multicast messages). Messages in a batch can be OOB messages, regular messages, or mixed
 * messages, although the transport itself will create initial MessageBatches that contain only either OOB or
 * regular messages.<p/>
 * The default processing below sends messages up the stack individually, based on a matching criteria
 * (calling {@link #accept(org.jgroups.Message)}), and - if true - calls {@link #up(org.jgroups.Event)}
 * for that message and removes the message. If the batch is not empty, it is passed up, or else it is dropped.<p/>
 * Subclasses should check if there are any messages destined for them (e.g. using
 * {@link MessageBatch#getMatchingMessages(short,boolean)}), then possibly remove and process them and finally pass
 * the batch up to the next protocol. Protocols can also modify messages in place, e.g. ENCRYPT could decrypt all
 * encrypted messages in the batch, not remove them, and pass the batch up when done.
 * @param batch The message batch
 */
public void up(MessageBatch batch) {
    for(Iterator<Message> it=batch.iterator(); it.hasNext();) {
        Message msg=it.next();
        if(msg != null && accept(msg)) {
            it.remove();
            try {
                up(new Event(Event.MSG, msg));
            }
            catch(Throwable t) {
                log.error(Util.getMessage("PassUpFailure"), t);
            }
        }
    }
    if(!batch.isEmpty())
        up_prot.up(batch);
}
 

import org.jgroups.Event; //导入依赖的package包/类
protected Object handleDownMessage(final Event evt, final Message msg, Address dest, int length) {
    if(dest != null) // 2nd line of defense, not really needed
        return down_prot.down(evt);

    long block_time=max_block_times != null? getMaxBlockTime(length) : max_block_time;
    while(running) {
        boolean rc=credits.decrement(length, block_time);
        if(rc || max_block_times != null || !running)
            break;

        if(needToSendCreditRequest()) {
            List<Tuple<Address,Long>> targets=credits.getMembersWithCreditsLessThan(min_credits);
            for(Tuple<Address,Long> tuple: targets)
                sendCreditRequest(tuple.getVal1(), Math.min(max_credits, max_credits - tuple.getVal2()));
        }
    }
    
    // send message - either after regular processing, or after blocking (when enough credits are available again)
    return down_prot.down(evt);
}
 

import org.jgroups.Event; //导入依赖的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.Event; //导入依赖的package包/类
/** Check if I'm in memberList, too */
/*public void start() throws Exception {
    super.start();
    IpAddress self=(IpAddress)auth.getPhysicalAddress();
    if(!isInMembersList(self))
        throw new IllegalStateException("own physical address " + self + " is not in members (" + memberList + ")");
} */

public boolean authenticate(AuthToken token, Message msg) {
    if ((token != null) && (token instanceof FixedMembershipToken) && (this.memberList != null)) {
        PhysicalAddress src = (PhysicalAddress) auth.down(new Event(Event.GET_PHYSICAL_ADDRESS, msg.getSrc()));
        if (src == null) {
            log.error("didn't find physical address for " + msg.getSrc());
            return false;
        }
        return isInMembersList((IpAddress)src);
    }

    if (log.isWarnEnabled())
        log.warn("Invalid AuthToken instance - wrong type or null");
    return false;
}
 

import org.jgroups.Event; //导入依赖的package包/类
public Object up(Event evt) {
    if(!up)
        return up_prot.up(evt);
    if(evt.getType() != Event.MSG)
        return up_prot.up(evt);
    Message msg=(Message)evt.getArg();
    synchronized(up_msgs) {
        up_msgs.add(msg);
    }
    if(up_msgs.size() >= max_size) {
        shuffleAndSendMessages();
    }
    else
        startTask();
    return null;
}
 

import org.jgroups.Event; //导入依赖的package包/类
/**
 * If there is no header, we pass the message up. Otherwise we uncompress the payload to its original size.
 * @param evt
 */
public Object up(Event evt) {
    if(evt.getType() == Event.MSG) {
        Message msg=(Message)evt.getArg();
        CompressHeader hdr=(CompressHeader)msg.getHeader(this.id);
        if(hdr != null) {
            Message uncompressed_msg=uncompress(msg, hdr.original_size);
            if(uncompressed_msg != null) {
                if(log.isTraceEnabled())
                    log.trace("up(): uncompressed " + msg.getLength() + " bytes to " + uncompressed_msg.getLength() + " bytes");
                return up_prot.up(new Event(Event.MSG, uncompressed_msg));
            }
        }
    }
    return up_prot.up(evt);
}
 

import org.jgroups.Event; //导入依赖的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.Event; //导入依赖的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
}