You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2015/07/27 22:27:56 UTC
[12/14] incubator-geode git commit: jgroups source is now removed
from asf branch GEODE-77 and is a downloaded dependency. Able to start/stop a
locator and a server via gfsh but there is no HA or authentication in the
membership system.
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java
new file mode 100755
index 0000000..902e1f6
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java
@@ -0,0 +1,325 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.locator;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.logging.log4j.Logger;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.cache.GemFireCache;
+import com.gemstone.gemfire.distributed.DistributedSystem;
+import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.distributed.internal.SharedConfiguration;
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.distributed.internal.membership.MembershipManager;
+import com.gemstone.gemfire.distributed.internal.membership.NetView;
+import com.gemstone.gemfire.distributed.internal.membership.gms.GMSUtil;
+import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
+import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.Locator;
+import com.gemstone.gemfire.distributed.internal.membership.gms.mgr.GMSMembershipManager;
+import com.gemstone.gemfire.distributed.internal.tcpserver.TcpClient;
+import com.gemstone.gemfire.distributed.internal.tcpserver.TcpHandler;
+import com.gemstone.gemfire.distributed.internal.tcpserver.TcpServer;
+import com.gemstone.gemfire.internal.Version;
+import com.gemstone.gemfire.internal.VersionedObjectInput;
+import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.logging.LogService;
+
+public class GMSLocator implements TcpHandler, Locator {
+
+ private static final int LOCATOR_FILE_STAMP = 0x7b8cf741;
+
+ private static final Logger logger = LogService.getLogger();
+
+ private final int port;
+ private final InetAddress bindAddress;
+ private final boolean usePreferredCoordinators;
+ private final boolean networkPartitionDetectionEnabled;
+ private final String locatorString;
+ private final List<InetSocketAddress> locators;
+ private Services services;
+
+ private Set<InternalDistributedMember> registrants = new HashSet<InternalDistributedMember>();
+
+ private NetView view;
+
+ private File viewFile;
+
+ /**
+ * @param mgr the membership manager
+ * @param port the tcp/ip server socket port number
+ * @param bindAddress network address to bind to
+ * @param stateFile the file to persist state to/recover from
+ * @param locatorString location of other locators (bootstrapping, failover)
+ * @param usePreferredCoordinators true if the membership coordinator should be a Locator
+ * @param networkPartitionDetectionEnabled true if network partition detection is enabled
+ */
+ public GMSLocator(int port,
+ InetAddress bindAddress,
+ File stateFile,
+ String locatorString,
+ boolean usePreferredCoordinators,
+ boolean networkPartitionDetectionEnabled) {
+ this.port = port;
+ this.bindAddress = bindAddress;
+ this.usePreferredCoordinators = usePreferredCoordinators;
+ this.networkPartitionDetectionEnabled = networkPartitionDetectionEnabled;
+ this.locatorString = locatorString;
+ if (this.locatorString == null || this.locatorString.length() == 0) {
+ this.locators = new ArrayList<InetSocketAddress>(0);
+ } else {
+ this.locators = GMSUtil.parseLocators(locatorString, bindAddress);
+ }
+ this.viewFile = stateFile;
+ }
+
+ /**
+ * This must be called after booting the membership manager so
+ * that the locator can use its services
+ * @param svc
+ */
+ public void setMembershipManager(MembershipManager mgr) {
+ logger.info("Peer locator is connecting to local membership services");
+ services = ((GMSMembershipManager)mgr).getServices();
+ services.setLocator(this);
+ this.view = services.getJoinLeave().getView();
+ }
+
+ public void init(TcpServer server) {
+ recover();
+ }
+
+ private void findServices() {
+ InternalDistributedSystem sys = InternalDistributedSystem.getAnyInstance();
+ if (sys != null && services == null) {
+ logger.info("Peer locator found distributed system " + sys);
+ setMembershipManager(sys.getDM().getMembershipManager());
+ }
+ }
+
+ public void installView(NetView view) {
+ synchronized(this.registrants) {
+ registrants.clear();
+ }
+ logger.info("Peer locator received new membership view: " + view);
+ this.view = view;
+ saveView(view);
+ }
+
+
+ @Override
+ public Object processRequest(Object request) throws IOException {
+ Object response = null;
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Peer locator processing " + request);
+ }
+
+ if (request instanceof GetViewRequest) {
+ if (view != null) {
+ response = new GetViewResponse(view);
+ }
+ } else if (request instanceof FindCoordinatorRequest) {
+ FindCoordinatorRequest findRequest = (FindCoordinatorRequest)request;
+
+ if (findRequest.getMemberID() != null) {
+ InternalDistributedMember coord = null;
+
+ // at this level we want to return the coordinator known to membership services,
+ // which may be more up-to-date than the one known by the membership manager
+ if (view == null) {
+ findServices();
+ }
+ if (view != null) {
+ coord = view.getCoordinator();
+ }
+
+ if (coord != null) {
+ // no need to keep track of registrants after we're in the distributed system
+ synchronized(registrants) {
+ registrants.clear();
+ }
+
+ } else {
+ // find the "oldest" registrant
+ synchronized(registrants) {
+ registrants.add(findRequest.getMemberID());
+ if (services != null) {
+ coord = services.getJoinLeave().getMemberID();
+ }
+ for (InternalDistributedMember mbr: registrants) {
+ if (mbr != coord && (coord==null || mbr.compareTo(coord) < 0)) {
+ coord = mbr;
+ }
+ }
+ }
+ }
+ response = new FindCoordinatorResponse(coord,
+ this.networkPartitionDetectionEnabled, this.usePreferredCoordinators);
+ }
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("Peer locator returning " + response);
+ }
+ return response;
+ }
+
+ public void saveView(NetView view) {
+ if (viewFile == null) {
+ return;
+ }
+ if (!viewFile.delete() && viewFile.exists()) {
+ logger.warn("Peer locator is unable to delete persistent membership information in " +
+ viewFile.getAbsolutePath());
+ }
+ try {
+ ObjectOutputStream oos = null;
+ try {
+ oos = new ObjectOutputStream(new FileOutputStream(viewFile));
+ oos.writeInt(LOCATOR_FILE_STAMP);
+ oos.writeInt(Version.CURRENT_ORDINAL);
+ DataSerializer.writeObject(view, oos);
+ }
+ finally {
+ oos.flush();
+ oos.close();
+ }
+ }
+ catch (Exception e) {
+ logger.warn("Peer locator encountered an error writing current membership to disk. Disabling persistence. Care should be taken when bouncing this locator as it will not be able to recover knowledge of the running distributed system", e);
+ this.viewFile = null;
+ // TODO - force TcpServer to shut down if this happens instead of continuing to run
+ }
+ }
+
+
+ @Override
+ public void endRequest(Object request, long startTime) {
+ // place holder for statistics
+ }
+
+ @Override
+ public void endResponse(Object request, long startTime) {
+ // place holder for statistics
+ }
+
+ @Override
+ public void shutDown() {
+ }
+
+
+ // test hook
+ public List<InternalDistributedMember> getMembers() {
+ if (view != null) {
+ return new ArrayList<InternalDistributedMember>(view.getMembers());
+ } else {
+ synchronized(registrants) {
+ return new ArrayList<InternalDistributedMember>(registrants);
+ }
+ }
+ }
+
+ @Override
+ public void restarting(DistributedSystem ds, GemFireCache cache,
+ SharedConfiguration sharedConfig) {
+ setMembershipManager(((InternalDistributedSystem)ds).getDM().getMembershipManager());
+ }
+
+ public void recover() {
+ if (!recoverFromOthers()) {
+ recoverFromFile(viewFile);
+ }
+ }
+
+ private boolean recoverFromOthers() {
+ if (locators.isEmpty()) {
+ return false;
+ }
+
+ for (InetSocketAddress other: this.locators) {
+ logger.info("Peer locator attempting to get state from " + other);
+ if (recover(other)) {
+ return true;
+ }
+ } // for
+ return false;
+ }
+
+ private boolean recover(InetSocketAddress other) {
+ try {
+ Object response = TcpClient.requestToServer(other.getAddress(), other.getPort(),
+ new GetViewRequest(), 20000, true);
+ if (response != null && (response instanceof GetViewResponse)) {
+ this.view = ((GetViewResponse)response).getView();
+ logger.info("Peer locator recovered initial membership of {}", view);
+ return true;
+ }
+ } catch (IOException e) {
+ // ignore
+ } catch (ClassNotFoundException e) {
+ // hmm - odd response?
+ }
+ return false;
+ }
+
+ private boolean recoverFromFile(File file) {
+ if (!file.exists()) {
+ return false;
+ }
+ logger.info("Peer locator recovering from " + file.getAbsolutePath());
+ FileInputStream fis = null;
+ try {
+ fis = new FileInputStream(file);
+ ObjectInput ois = new ObjectInputStream(fis);
+
+ int magic = ois.readInt();
+ if (magic != LOCATOR_FILE_STAMP) {
+ return false;
+ }
+
+ int version = ois.readInt();
+ Version geodeVersion = Version.fromOrdinalNoThrow((short)version, false);
+ if (geodeVersion != null && version != Version.CURRENT_ORDINAL) {
+ logger.info("Peer locator found that persistent view was written with {}", geodeVersion);
+ ois = new VersionedObjectInput(ois, geodeVersion);
+ } else {
+ return false;
+ }
+
+ this.view = DataSerializer.readObject(ois);
+
+ logger.info("Initial membership is " + view);
+ return true;
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ logger.warn(LocalizedStrings.Locator_unable_to_recover_view_0
+ .toLocalizedString(file), e);
+ try {
+ fis.close();
+ } catch (IOException ex) {
+ // ignore
+ }
+ if (!file.delete() && file.exists()) {
+ logger.warn("Peer locator was unable to recover from or delete " + file);
+ this.viewFile = null;
+ }
+ }
+ finally {
+ try { fis.close(); } catch (IOException e) { }
+ }
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewRequest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewRequest.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewRequest.java
new file mode 100755
index 0000000..20ababe
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewRequest.java
@@ -0,0 +1,33 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.locator;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.internal.DataSerializableFixedID;
+import com.gemstone.gemfire.internal.Version;
+
+public class GetViewRequest implements DataSerializableFixedID, PeerLocatorRequest {
+
+ public GetViewRequest() {
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+
+ @Override
+ public int getDSFID() {
+ return GET_VIEW_REQ;
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewResponse.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewResponse.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewResponse.java
new file mode 100755
index 0000000..5a6eab9
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GetViewResponse.java
@@ -0,0 +1,48 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.locator;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.distributed.internal.membership.NetView;
+import com.gemstone.gemfire.internal.DataSerializableFixedID;
+import com.gemstone.gemfire.internal.Version;
+
+public class GetViewResponse implements DataSerializableFixedID {
+
+ private NetView view;
+
+ public GetViewResponse(NetView view) {
+ this.view = view;
+ }
+
+ public GetViewResponse() {
+ // no-arg constructor for serialization
+ }
+
+ public NetView getView() {
+ return view;
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+
+ @Override
+ public int getDSFID() {
+ return GET_VIEW_RESP;
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ DataSerializer.writeObject(view, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ view = DataSerializer.readObject(in);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/PeerLocatorRequest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/PeerLocatorRequest.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/PeerLocatorRequest.java
new file mode 100755
index 0000000..20be4f1
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/PeerLocatorRequest.java
@@ -0,0 +1,9 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.locator;
+
+/**
+ * PeerLocatorRequest is a marker for messages that are intended to be handled
+ * by the Peer location service in the Geode Locator.
+ */
+public interface PeerLocatorRequest {
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
new file mode 100755
index 0000000..f94d9c4
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
@@ -0,0 +1,909 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.membership;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.logging.log4j.Logger;
+
+import com.gemstone.gemfire.CancelCriterion;
+import com.gemstone.gemfire.SystemConnectException;
+import com.gemstone.gemfire.distributed.DistributedMember;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.DistributionMessage;
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.distributed.internal.membership.NetView;
+import com.gemstone.gemfire.distributed.internal.membership.gms.GMSMember;
+import com.gemstone.gemfire.distributed.internal.membership.gms.GMSUtil;
+import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
+import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.JoinLeave;
+import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.MessageHandler;
+import com.gemstone.gemfire.distributed.internal.membership.gms.locator.FindCoordinatorRequest;
+import com.gemstone.gemfire.distributed.internal.membership.gms.locator.FindCoordinatorResponse;
+import com.gemstone.gemfire.distributed.internal.membership.gms.messages.InstallViewMessage;
+import com.gemstone.gemfire.distributed.internal.membership.gms.messages.JoinRequestMessage;
+import com.gemstone.gemfire.distributed.internal.membership.gms.messages.JoinResponseMessage;
+import com.gemstone.gemfire.distributed.internal.membership.gms.messages.LeaveRequestMessage;
+import com.gemstone.gemfire.distributed.internal.membership.gms.messages.RemoveMemberMessage;
+import com.gemstone.gemfire.distributed.internal.membership.gms.messages.ViewAckMessage;
+import com.gemstone.gemfire.distributed.internal.tcpserver.TcpClient;
+import com.gemstone.gemfire.internal.logging.LoggingThreadGroup;
+import com.gemstone.gemfire.security.AuthenticationFailedException;
+
+/**
+ * GMSJoinLeave handles membership communication with other processes in the
+ * distributed system. It replaces the JGroups channel membership services
+ * that Geode formerly used for this purpose.
+ *
+ */
+public class GMSJoinLeave implements JoinLeave, MessageHandler {
+
+ /** number of times to try joining before giving up */
+ private static final int JOIN_ATTEMPTS = Integer.getInteger("geode.join-attempts", 6);
+
+ /** amount of time to sleep before trying to join after a failed attempt */
+ private static final int JOIN_RETRY_SLEEP = Integer.getInteger("geode.join-retry-sleep", 3000);
+
+ /** amount of time to wait for a view to be acked by all members before performing suspect processing on non-responders */
+ private static final int VIEW_INSTALLATION_TIMEOUT = Integer.getInteger("geode.view-ack-timeout", 12500);
+
+ /** stall time to wait for concurrent join/leave/remove requests to be received */
+ private static final long MEMBER_REQUEST_COLLECTION_INTERVAL = Long.getLong("geode.member-request-collection-interval", 5000);
+
+ /** membership logger */
+ private static final Logger logger = Services.getLogger();
+
+
+ /** the view ID where I entered into membership */
+ private int birthViewId;
+
+ /** my address */
+ private InternalDistributedMember localAddress;
+
+ private Services services;
+
+ private boolean isConnected;
+
+ /** a lock governing GMS state */
+ private ReadWriteLock stateLock = new ReentrantReadWriteLock();
+
+ /** guarded by stateLock */
+ private boolean isCoordinator;
+
+ private final Object viewInstallationLock = new Object();
+
+ /** the currently installed view */
+ private volatile NetView currentView;
+
+ /** a new view being installed */
+ private NetView preparedView;
+
+ private List<InetSocketAddress> locators;
+
+ private InternalDistributedMember leader;
+
+ /** a list of join/leave/crashes */
+ private final List<DistributionMessage> viewRequests = new LinkedList<DistributionMessage>();
+
+ /** collects the response to a join request */
+ private JoinResponseMessage[] joinResponse = new JoinResponseMessage[1];
+
+ private ViewReplyProcessor viewResponses = new ViewReplyProcessor();
+
+ private boolean disableForcedDisconnect = false;
+
+ /** background thread that creates new membership views */
+ private ViewCreator viewCreator;
+
+ private volatile boolean isStopping;
+
+
+ /**
+ * attempt to join the distributed system
+ * loop
+ * send a join request to a locator & get a response
+ *
+ * If the response indicates there's no coordinator it
+ * will contain a set of members that have recently contacted
+ * it. The "oldest" member is selected as the coordinator
+ * based on ID sort order.
+ *
+ * @return true if successful, false if not
+ */
+ public boolean join() {
+ for (int tries=0; tries<JOIN_ATTEMPTS; tries++) {
+ InternalDistributedMember coord = findCoordinator();
+ logger.debug("found possible coordinator {}", coord);
+ if (coord != null) {
+ if (coord.equals(this.localAddress)) {
+ if (tries > (JOIN_ATTEMPTS/2)) {
+ becomeCoordinator();
+ return true;
+ }
+ } else {
+ if (attemptToJoin(coord)) {
+ return true;
+ }
+ }
+ }
+ try {
+ Thread.sleep(JOIN_RETRY_SLEEP);
+ } catch (InterruptedException e) {
+ return false;
+ }
+ } // for
+ return this.isConnected;
+ }
+
+ /**
+ * send a join request and wait for a reply. Process the reply.
+ * This may throw a SystemConnectException or an AuthenticationFailedException
+ * @param coord
+ * @return true if the attempt succeeded, false if it timed out
+ */
+ private boolean attemptToJoin(InternalDistributedMember coord) {
+ // send a join request to the coordinator and wait for a response
+ logger.info("Attempting to join the distributed system through coordinator " + coord + " using address " + this.localAddress);
+ JoinRequestMessage req = new JoinRequestMessage(coord, this.localAddress,
+ services.getAuthenticator().getCredentials());
+ try {
+ services.getMessenger().send(req);
+ } catch (IOException e) {
+ throw new SystemConnectException("Exception caught while trying to join", e);
+ }
+ JoinResponseMessage response = null;
+ synchronized(joinResponse) {
+ if (joinResponse[0] == null) {
+ try {
+ joinResponse.wait(services.getConfig().getJoinTimeout()/JOIN_ATTEMPTS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ response = joinResponse[0];
+ }
+ if (response != null) {
+ joinResponse[0] = null;
+ String failReason = response.getRejectionMessage();
+ if (failReason != null) {
+ if (failReason.contains("Rejecting the attempt of a member using an older version")
+ || failReason.contains("15806")) {
+ throw new SystemConnectException(failReason);
+ }
+ throw new AuthenticationFailedException(failReason);
+ }
+ if (response.getCurrentView() != null) {
+ this.birthViewId = response.getMemberID().getVmViewId();
+ this.localAddress.setVmViewId(this.birthViewId);
+ GMSMember me = (GMSMember)this.localAddress.getNetMember();
+ GMSMember o = (GMSMember)response.getMemberID().getNetMember();
+ me.setSplitBrainEnabled(o.isSplitBrainEnabled());
+ me.setShouldNotBeCoordinator(o.isShouldNotBeCoordinator());
+ installView(response.getCurrentView());
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * process a join request from another member. If this is the coordinator
+ * this method will enqueue the request for processing in another thread.
+ * If this is not the coordinator but the coordinator is known, the message
+ * is forwarded to the coordinator.
+ * @param incomingRequest
+ */
+ private void processJoinRequest(JoinRequestMessage incomingRequest) {
+ NetView v = currentView;
+ if (!isCoordinator && v != null) {
+ // forward the request to the coordinator
+ logger.info("forwarding join request from " + incomingRequest.getMemberID() + " to " + v.getCoordinator());
+ incomingRequest.setRecipient(v.getCoordinator());
+ try {
+ services.getMessenger().send(incomingRequest);
+ } catch (IOException ignored) {
+ // this isn't possible since this was an incoming message
+ }
+ } else {
+ Object creds = incomingRequest.getCredentials();
+ if (creds != null) {
+ String rejection = null;
+ try {
+ rejection = services.getAuthenticator().authenticate(incomingRequest.getMemberID(), creds);
+ } catch (AuthenticationFailedException e) {
+ rejection = e.getMessage();
+ }
+ if (rejection != null && rejection.length() > 0) {
+ JoinResponseMessage m = new JoinResponseMessage(rejection);
+ m.setRecipient(incomingRequest.getMemberID());
+ try {
+ services.getMessenger().send(m);
+ } catch (IOException e2) {
+ logger.info("unable to send join response " + rejection + " to " + incomingRequest.getMemberID(), e2);
+ }
+ }
+ }
+ recordViewRequest(incomingRequest);
+ }
+ }
+
+
+ /**
+ * Process a Leave request from another member. This may cause this member
+ * to become the new membership coordinator. If this is the coordinator
+ * a new view will be triggered.
+ *
+ * @param incomingRequest
+ */
+ private void processLeaveRequest(LeaveRequestMessage incomingRequest) {
+ NetView v = currentView;
+ if (logger.isDebugEnabled()) {
+ logger.debug("JoinLeave.processLeaveRequest invoked. isCoordinator="+isCoordinator+ "; isStopping="+isStopping
+ +"; cancelInProgress="+services.getCancelCriterion().isCancelInProgress());
+ }
+ if (!isCoordinator && !isStopping && !services.getCancelCriterion().isCancelInProgress()) {
+ logger.debug("JoinLeave is checking to see if I should become coordinator");
+ NetView check = new NetView(v, v.getViewId()+1);
+ check.remove(incomingRequest.getMemberID());
+ if (check.getCoordinator().equals(localAddress)) {
+ becomeCoordinator(incomingRequest.getMemberID());
+ }
+ }
+ else {
+ if (!isStopping && !services.getCancelCriterion().isCancelInProgress()) {
+ recordViewRequest(incomingRequest);
+ }
+ }
+ }
+
+
+ /**
+ * Process a Remove request from another member. This may cause this member
+ * to become the new membership coordinator. If this is the coordinator
+ * a new view will be triggered.
+ *
+ * @param incomingRequest
+ */
+ private void processRemoveRequest(RemoveMemberMessage incomingRequest) {
+ NetView v = currentView;
+ if (logger.isDebugEnabled()) {
+ logger.debug("JoinLeave.processRemoveRequest invoked. isCoordinator="+isCoordinator+ "; isStopping="+isStopping
+ +"; cancelInProgress="+services.getCancelCriterion().isCancelInProgress());
+ }
+ InternalDistributedMember mbr = incomingRequest.getMemberID();
+
+ if (v != null && !v.contains(incomingRequest.getSender())) {
+ logger.info("Membership ignoring removal request for " + mbr + " from non-member " + incomingRequest.getSender());
+ return;
+ }
+
+ logger.info("Membership received a request to remove " + mbr
+ + "; reason="+incomingRequest.getReason());
+
+ if (mbr.equals(this.localAddress)) {
+ // oops - I've been kicked out
+ services.getManager().forceDisconnect(incomingRequest.getReason());
+ return;
+ }
+
+ if (!isCoordinator && !isStopping && !services.getCancelCriterion().isCancelInProgress()) {
+ logger.debug("JoinLeave is checking to see if I should become coordinator");
+ NetView check = new NetView(v, v.getViewId()+1);
+ check.remove(mbr);
+ if (check.getCoordinator().equals(localAddress)) {
+ becomeCoordinator(mbr);
+ }
+ }
+ else {
+ if (!isStopping && !services.getCancelCriterion().isCancelInProgress()) {
+ recordViewRequest(incomingRequest);
+ }
+ }
+ }
+
+
+ private void recordViewRequest(DistributionMessage request) {
+ logger.debug("JoinLeave is recording the request to be processed in the next membership view");
+ synchronized(viewRequests) {
+ viewRequests.add(request);
+ viewRequests.notify();
+ }
+ startCoordinatorServices();
+ }
+
+
+ /**
+ * Yippeee - I get to be the coordinator
+ */
+ private void becomeCoordinator() {
+ becomeCoordinator(null);
+ }
+
+ /**
+ * @param oldCoordinator may be null
+ */
+ private void becomeCoordinator(InternalDistributedMember oldCoordinator) {
+ stateLock.writeLock().lock();
+ try {
+ if (isCoordinator) {
+ return;
+ }
+ logger.debug("JoinLeave: this member is becoming the membership coordinator with address {}", localAddress);
+ isCoordinator = true;
+ if (currentView == null) {
+ // create the initial membership view
+ NetView newView = new NetView(this.localAddress);
+ this.localAddress.setVmViewId(0);
+ installView(newView);
+ isConnected = true;
+ startCoordinatorServices();
+ } else {
+ // create and send out a new view
+ NetView newView;
+ synchronized(viewInstallationLock) {
+ int viewNumber = currentView.getViewId() + 5;
+ startCoordinatorServices();
+ List<InternalDistributedMember> mbrs = new ArrayList<InternalDistributedMember>(currentView.getMembers());
+ mbrs.add(localAddress);
+ List<InternalDistributedMember> leaving = new ArrayList<InternalDistributedMember>();
+ if (oldCoordinator != null) {
+ leaving.add(oldCoordinator);
+ }
+ newView = new NetView(this.localAddress, viewNumber, mbrs, leaving,
+ Collections.EMPTY_LIST);
+ }
+ sendView(newView);
+ }
+ } finally {
+ stateLock.writeLock().unlock();
+ }
+ }
+
+
+ private void sendJoinResponses(List<InternalDistributedMember> newMbrs, NetView newView) {
+ for (InternalDistributedMember mbr: newMbrs) {
+ JoinResponseMessage response = new JoinResponseMessage(mbr, newView);
+ try {
+ services.getMessenger().send(response);
+ } catch (IOException e) {
+ logger.info("unable to send join response to {}", mbr);
+ }
+ }
+ }
+
+ private void sendRemoveMessages(List<InternalDistributedMember> newMbrs,
+ List<String> reasons, NetView newView) {
+ Iterator<String> reason = reasons.iterator();
+ for (InternalDistributedMember mbr: newMbrs) {
+ RemoveMemberMessage response = new RemoveMemberMessage(mbr, mbr, reason.next());
+ try {
+ services.getMessenger().send(response);
+ } catch (IOException e) {
+ logger.info("unable to send remove message to {}", mbr);
+ }
+ }
+ }
+
+ void sendView(NetView view) {
+ // TODO two-phase view installation and network partition detection
+ // send a prepared view, check for conflicting prepared view
+ // examine members that don't respond & recreate view and start over
+ int id = view.getViewId();
+ InstallViewMessage msg = new InstallViewMessage(view, services.getAuthenticator().getCredentials());
+ Set<InternalDistributedMember> recips = new HashSet<InternalDistributedMember>(view.getMembers());
+ recips.addAll(view.getCrashedMembers());
+ msg.setRecipients(recips);
+ this.viewResponses.initialize(id, recips);
+ logger.info("View Creator sending new view " + view);
+ try {
+ services.getMessenger().send(msg);
+ }
+ catch (IOException e) {
+ logger.warn("Unsuccessful in installing new membership view", e);
+ }
+
+ Set<InternalDistributedMember> failedToRespond = this.viewResponses.waitForResponses();
+
+ logger.info("View Creator is finished waiting for responses to view change");
+
+ if (!failedToRespond.isEmpty() && (services.getCancelCriterion().cancelInProgress() == null)) {
+ logger.warn("these members failed to respond to the view change: " + failedToRespond);
+ }
+ }
+
+
+
+ private void processViewMessage(InstallViewMessage m) {
+ NetView view = m.getView();
+ if (currentView != null && view.getViewId() < currentView.getViewId()) {
+ // ignore old views
+ ackView(m);
+ return;
+ }
+
+ if (currentView != null && !m.isPreparing() && !view.contains(this.localAddress)) {
+ services.getManager().forceDisconnect("I am no longer in the membership view");
+ return;
+ }
+
+ if (m.isPreparing()) {
+ if (this.preparedView != null && this.preparedView.getViewId() >= view.getViewId()) {
+ try {
+ services.getMessenger().send(new ViewAckMessage(m.getSender(), this.preparedView));
+ } catch (IOException e) {
+ logger.info("unable to send view response to " + m.getSender(), e);
+ }
+ }
+ }
+
+ if (view.contains(view.getCreator())) {
+ ackView(m);
+ }
+
+ // process the view
+ installView(view);
+ }
+
+
+ private void ackView(InstallViewMessage m) {
+ // send an acknowledgement
+ try {
+ services.getMessenger().send(new ViewAckMessage(m.getSender(), m.getView().getViewId()));
+ } catch (IOException e) {
+ logger.info("unable to send view response to " + m.getSender(), e);
+ }
+ }
+
+
+ private void processViewAckMessage(ViewAckMessage m) {
+ this.viewResponses.processViewResponse(m.getViewId(), m.getSender());
+ }
+ /**
+ * This contacts the locators to find out who the current coordinator is.
+ * All locators are contacted. If they don't agree then we choose the oldest
+ * coordinator and return it.
+ * @return
+ */
+ private InternalDistributedMember findCoordinator() {
+ if (locators == null) {
+ DistributionConfig dconfig = services.getConfig().getDistributionConfig();
+ String bindAddr = dconfig.getBindAddress();
+ locators = GMSUtil.parseLocators(dconfig.getLocators(), bindAddr);
+ }
+ assert this.localAddress != null;
+ FindCoordinatorRequest request = new FindCoordinatorRequest(this.localAddress);
+ Set<InternalDistributedMember> coordinators = new HashSet<InternalDistributedMember>();
+ for (InetSocketAddress addr: locators) {
+ try {
+ Object o = TcpClient.requestToServer(
+ addr.getAddress(), addr.getPort(), request, services.getConfig().getJoinTimeout(),
+ true);
+ FindCoordinatorResponse response = (o instanceof FindCoordinatorResponse) ? (FindCoordinatorResponse)o : null;
+ if (response != null && response.getCoordinator() != null) {
+ coordinators.add(response.getCoordinator());
+ GMSMember mbr = (GMSMember)this.localAddress.getNetMember();
+ services.getConfig().setNetworkPartitionDetectionEnabled(response.isNetworkPartitionDetectionEnabled());
+ if (response.isUsePreferredCoordinators()
+ && localAddress.getVmKind() != DistributionManager.LOCATOR_DM_TYPE) {
+ mbr.setShouldNotBeCoordinator(true);
+ }
+ }
+ } catch (IOException problem) {
+ } catch (ClassNotFoundException problem) {
+ }
+ }
+ if (coordinators.isEmpty()) {
+ return null;
+ }
+ Iterator<InternalDistributedMember> it = coordinators.iterator();
+ if (coordinators.size() == 1) {
+ return it.next();
+ }
+ InternalDistributedMember oldest = it.next();
+ while (it.hasNext()) {
+ InternalDistributedMember candidate = it.next();
+ if (oldest.compareTo(candidate) > 0) {
+ oldest = candidate;
+ }
+ }
+ return oldest;
+ }
+
+
+ /**
+ * receives a JoinResponse holding a membership view or rejection message
+ * @param rsp
+ */
+ private void processJoinResponse(JoinResponseMessage rsp) {
+ synchronized(joinResponse) {
+ joinResponse[0] = rsp;
+ joinResponse.notify();
+ }
+ }
+
+ @Override
+ public NetView getView() {
+ return currentView;
+ }
+
+ @Override
+ public InternalDistributedMember getMemberID() {
+ return this.localAddress;
+ }
+
+ public void installView(NetView newView) {
+ synchronized(viewInstallationLock) {
+ if (currentView != null && currentView.getViewId() >= newView.getViewId()) {
+ // old view - ignore it
+ return;
+ }
+ checkForPartition(newView);
+ currentView = newView;
+ leader = newView.getLeadMember();
+ preparedView = null;
+ services.installView(newView);
+ if (!newView.getCreator().equals(this.localAddress)) {
+ if (newView.shouldBeCoordinator(this.localAddress)) {
+ becomeCoordinator();
+ } else if (this.isCoordinator) {
+ // stop being coordinator
+ stateLock.writeLock().lock();
+ try {
+ stopCoordinatorServices();
+ this.isCoordinator = false;
+ } finally {
+ stateLock.writeLock().unlock();
+ }
+ }
+ }
+ }
+ }
+
+
+ private void checkForPartition(NetView newView) {
+ // compare currentView to newView to see if there's been a network
+ // partition event
+
+ }
+
+
+ /** invoke this under the viewInstallationLock */
+ private void startCoordinatorServices() {
+ if (viewCreator == null || viewCreator.isShutdown()) {
+ viewCreator = new ViewCreator("Geode Membership View Creator", Services.getThreadGroup());
+ viewCreator.setDaemon(true);
+ viewCreator.start();
+ }
+ }
+
+ private void stopCoordinatorServices() {
+ if (viewCreator != null && !viewCreator.isShutdown()) {
+ viewCreator.shutdown();
+ }
+ }
+
+ public static void loadEmergencyClasses() {
+ }
+
+ @Override
+ public void emergencyClose() {
+ isStopping = true;
+ isConnected = false;
+ stopCoordinatorServices();
+ isCoordinator = false;
+ currentView = null;
+ }
+
+ public void beSick() {
+ }
+
+ public void playDead() {
+ }
+
+ public void beHealthy() {
+ }
+
+
+
+ @Override
+ public void start() {
+ }
+
+
+
+ @Override
+ public void started() {
+ this.localAddress = services.getMessenger().getMemberID();
+ }
+
+
+
+ @Override
+ public void stop() {
+ logger.debug("JoinLeave stopping");
+ leave();
+ }
+
+
+
+ @Override
+ public void stopped() {
+ // TODO Auto-generated method stub
+
+ }
+
+
+
+ @Override
+ public void leave() {
+ synchronized(viewInstallationLock) {
+ NetView view = currentView;
+ isStopping = true;
+ if (view != null) {
+ if (view.size() > 1) {
+ if (this.isCoordinator) {
+ logger.debug("JoinLeave stopping coordination services");
+ stopCoordinatorServices();
+ NetView newView = new NetView(view, view.getViewId()+1);
+ newView.remove(localAddress);
+ InstallViewMessage m = new InstallViewMessage(newView, services.getAuthenticator().getCredentials());
+ m.setRecipients(newView.getMembers());
+ try {
+ services.getMessenger().send(m);
+ } catch (IOException e) {
+ logger.info("JoinLeave: unable to notify remaining members shutdown due to i/o exception", e);
+ }
+ }
+ else {
+ logger.debug("JoinLeave sending a leave request to {}", view.getCoordinator());
+ LeaveRequestMessage m = new LeaveRequestMessage(view.getCoordinator(), this.localAddress);
+ try {
+ services.getMessenger().send(m);
+ } catch (IOException e) {
+ logger.info("JoinLeave: unable to notify membership coordinator of shutdown due to i/o exception", e);
+ }
+ }
+ } // view.size
+ }// view != null
+ }
+
+ }
+
+
+
+ @Override
+ public void remove(InternalDistributedMember m, String reason) {
+ NetView v = this.currentView;
+ if (v != null) {
+ RemoveMemberMessage msg = new RemoveMemberMessage(v.getCoordinator(), m,
+ reason);
+ try {
+ services.getMessenger().send(msg);
+ } catch (IOException e) {
+ logger.info("JoinLeave was unable to remove member " + m + " due to an i/o exception");
+ }
+ }
+ }
+
+ @Override
+ public void disableDisconnectOnQuorumLossForTesting() {
+ this.disableForcedDisconnect = true;
+ }
+
+ @Override
+ public void init(Services s) {
+ this.services = s;
+ services.getMessenger().addHandler(JoinRequestMessage.class, this);
+ services.getMessenger().addHandler(JoinResponseMessage.class, this);
+ services.getMessenger().addHandler(InstallViewMessage.class, this);
+ services.getMessenger().addHandler(ViewAckMessage.class, this);
+ services.getMessenger().addHandler(LeaveRequestMessage.class, this);
+ services.getMessenger().addHandler(RemoveMemberMessage.class, this);
+
+ DistributionConfig dc = services.getConfig().getDistributionConfig();
+ int ackCollectionTimeout = dc.getMemberTimeout() * 2 * 12437 / 10000;
+ if (ackCollectionTimeout < 1500) {
+ ackCollectionTimeout = 1500;
+ } else if (ackCollectionTimeout > 12437) {
+ ackCollectionTimeout = 12437;
+ }
+ ackCollectionTimeout = Integer.getInteger("gemfire.VIEW_ACK_TIMEOUT", ackCollectionTimeout).intValue();
+
+ }
+
+ @Override
+ public void processMessage(DistributionMessage m) {
+ if (isStopping) {
+ return;
+ }
+ logger.debug("JoinLeave processing {}", m);
+ if (m instanceof JoinRequestMessage) {
+ processJoinRequest((JoinRequestMessage)m);
+ } else if (m instanceof JoinResponseMessage) {
+ processJoinResponse((JoinResponseMessage)m);
+ } else if (m instanceof InstallViewMessage) {
+ processViewMessage((InstallViewMessage)m);
+ } else if (m instanceof ViewAckMessage) {
+ processViewAckMessage((ViewAckMessage)m);
+ } else if (m instanceof LeaveRequestMessage) {
+ processLeaveRequest((LeaveRequestMessage)m);
+ } else if (m instanceof RemoveMemberMessage) {
+ processRemoveRequest((RemoveMemberMessage)m);
+ } else {
+ throw new IllegalArgumentException("unknown message type: " + m);
+ }
+ }
+
+
+
+
+ class ViewReplyProcessor {
+ volatile int viewId = -1;
+ volatile Set<InternalDistributedMember> recipients;
+
+ void initialize(int viewId, Set<InternalDistributedMember> recips) {
+ this.viewId = viewId;
+ this.recipients = recips;
+ }
+
+ void processViewResponse(int viewId, InternalDistributedMember sender) {
+ if (viewId == this.viewId) {
+ Set<InternalDistributedMember> waitingFor = this.recipients;
+ waitingFor.remove(sender);
+ if (waitingFor.isEmpty()) {
+ synchronized(waitingFor) {
+ waitingFor.notify();
+ }
+ }
+ }
+ }
+
+ Set<InternalDistributedMember> waitForResponses() {
+ Set<InternalDistributedMember> result = this.recipients;
+ long endOfWait = System.currentTimeMillis() + VIEW_INSTALLATION_TIMEOUT;
+ while (System.currentTimeMillis() < endOfWait
+ && (services.getCancelCriterion().cancelInProgress() == null)) {
+ try {
+ synchronized(result) {
+ result.wait(1000);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return result;
+ }
+ }
+ return result;
+ }
+ }
+
+
+
+
+ class ViewCreator extends Thread {
+ boolean shutdown = false;
+
+ ViewCreator(String name, ThreadGroup tg) {
+ super(tg, name);
+ }
+
+ void shutdown() {
+ shutdown = true;
+ synchronized(viewRequests) {
+ viewRequests.notify();
+ interrupt();
+ }
+ }
+
+ boolean isShutdown() {
+ return shutdown;
+ }
+
+ @Override
+ public void run() {
+ List<DistributionMessage> requests = null;
+ logger.info("View Creator thread is starting");
+ long okayToCreateView = System.currentTimeMillis() + MEMBER_REQUEST_COLLECTION_INTERVAL;
+ try {
+ for (;;) {
+ synchronized(viewRequests) {
+ if (shutdown) {
+ return;
+ }
+ if (viewRequests.isEmpty()) {
+ try {
+ logger.debug("View Creator is waiting for requests");
+ viewRequests.wait();
+ } catch (InterruptedException e) {
+ return;
+ }
+ } else {
+ if (System.currentTimeMillis() < okayToCreateView) {
+ // sleep to let more requests arrive
+ try {
+ sleep(100);
+ continue;
+ } catch (InterruptedException e) {
+ return;
+ }
+ } else {
+ // time to create a new membership view
+ requests = new ArrayList<DistributionMessage>(viewRequests);
+ viewRequests.clear();
+ okayToCreateView = System.currentTimeMillis() + MEMBER_REQUEST_COLLECTION_INTERVAL;
+ }
+ }
+ } // synchronized
+ if (requests != null && !requests.isEmpty()) {
+ logger.debug("View Creator is processing {} requests for the next membership view", requests.size());
+ createAndSendView(requests);
+ requests = null;
+ }
+ }
+ } finally {
+ shutdown = true;
+ }
+ }
+
+ void createAndSendView(List<DistributionMessage> requests) {
+ List<InternalDistributedMember> joinReqs = new ArrayList<InternalDistributedMember>();
+ List<InternalDistributedMember> leaveReqs = new ArrayList<InternalDistributedMember>();
+ List<InternalDistributedMember> removalReqs = new ArrayList<InternalDistributedMember>();
+ List<String> removalReasons = new ArrayList<String>();
+
+ for (DistributionMessage msg: requests) {
+ logger.debug("processing request {}", msg);
+ if (msg instanceof JoinRequestMessage) {
+ InternalDistributedMember mbr = ((JoinRequestMessage)msg).getMemberID();
+ joinReqs.add(mbr);
+ }
+ else if (msg instanceof LeaveRequestMessage) {
+ leaveReqs.add(((LeaveRequestMessage) msg).getMemberID());
+ }
+ else if (msg instanceof RemoveMemberMessage) {
+ removalReqs.add(((RemoveMemberMessage) msg).getMemberID());
+ removalReasons.add(((RemoveMemberMessage) msg).getReason());
+ }
+ else {
+ // TODO: handle removals
+ logger.warn("Unknown membership request encountered: {}", msg);
+ }
+ }
+
+ NetView newView;
+ synchronized(viewInstallationLock) {
+ int viewNumber = 0;
+ List<InternalDistributedMember> mbrs;
+ if (currentView == null) {
+ mbrs = new ArrayList<InternalDistributedMember>(joinReqs.size());
+ } else {
+ viewNumber = currentView.getViewId()+1;
+ mbrs = new ArrayList<InternalDistributedMember>(currentView.getMembers());
+ }
+ mbrs.addAll(joinReqs);
+ mbrs.removeAll(leaveReqs);
+ mbrs.removeAll(removalReqs);
+ newView = new NetView(localAddress, viewNumber, mbrs, leaveReqs,
+ removalReqs);
+ }
+ for (InternalDistributedMember mbr: joinReqs) {
+ mbr.setVmViewId(newView.getViewId());
+ }
+ // send removal messages before installing the view so we stop
+ // getting messages from them
+ sendRemoveMessages(removalReqs, removalReasons, newView);
+ sendJoinResponses(joinReqs, newView);
+ sendView(newView);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/InstallViewMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/InstallViewMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/InstallViewMessage.java
new file mode 100755
index 0000000..d8616d0
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/InstallViewMessage.java
@@ -0,0 +1,81 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messages;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.HighPriorityDistributionMessage;
+import com.gemstone.gemfire.distributed.internal.membership.NetView;
+import com.gemstone.gemfire.distributed.internal.membership.gms.mgr.GMSMembershipManager;
+
+public class InstallViewMessage extends HighPriorityDistributionMessage {
+
+ private NetView view;
+ private Object credentials;
+ private boolean preparing;
+
+ public InstallViewMessage(NetView view, Object credentials) {
+ this.view = view;
+ this.preparing = false;
+ this.credentials = credentials;
+ }
+
+ public InstallViewMessage(NetView view, Object credentials, boolean preparing) {
+ this.view = view;
+ this.preparing = preparing;
+ this.credentials = credentials;
+ }
+
+ public InstallViewMessage() {
+ // no-arg constructor for serialization
+ }
+
+ public NetView getView() {
+ return view;
+ }
+
+ public Object getCredentials() {
+ return credentials;
+ }
+
+ public boolean isPreparing() {
+ return preparing;
+ }
+
+ @Override
+ public int getDSFID() {
+ return INSTALL_VIEW_MESSAGE;
+ }
+
+ @Override
+ protected void process(DistributionManager dm) {
+ throw new IllegalStateException("this message is not intended to execute in a thread pool");
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ super.toData(out);
+ DataSerializer.writeObject(this.view, out);
+ DataSerializer.writeObject(this.credentials, out);
+ out.writeBoolean(preparing);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ super.fromData(in);
+ this.view = DataSerializer.readObject(in);
+ this.credentials = DataSerializer.readObject(in);
+ this.preparing = in.readBoolean();
+ }
+
+ @Override
+ public String toString() {
+ String s = getSender() == null? getRecipientsDescription() : ""+getSender();
+ return "InstallViewMessage("+s+"; preparing="+this.preparing+"; "+this.view
+ +"; cred="+(credentials==null?"null": "not null")
+ +")";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinRequestMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinRequestMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinRequestMessage.java
new file mode 100755
index 0000000..8621af8
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinRequestMessage.java
@@ -0,0 +1,69 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messages;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.HighPriorityDistributionMessage;
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.internal.Version;
+
+public class JoinRequestMessage extends HighPriorityDistributionMessage {
+ private InternalDistributedMember memberID;
+ private Object credentials;
+
+
+ public JoinRequestMessage(InternalDistributedMember coord, InternalDistributedMember id, Object credentials) {
+ super();
+ setRecipient(coord);
+ this.memberID = id;
+ this.credentials = credentials;
+ }
+
+ public JoinRequestMessage() {
+ // no-arg constructor for serialization
+ }
+
+ @Override
+ public int getDSFID() {
+ return JOIN_REQUEST;
+ }
+
+ @Override
+ public void process(DistributionManager dm) {
+ throw new IllegalStateException("this message is not intended to execute in a thread pool");
+ }
+
+ public InternalDistributedMember getMemberID() {
+ return memberID;
+ }
+
+ public Object getCredentials() {
+ return credentials;
+ }
+
+ @Override
+ public String toString() {
+ return getShortClassName() + "(" + memberID + (credentials==null? ")" : "; with credentials)");
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ DataSerializer.writeObject(memberID, out);
+ DataSerializer.writeObject(credentials, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ memberID = DataSerializer.readObject(in);
+ credentials = DataSerializer.readObject(in);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinResponseMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinResponseMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinResponseMessage.java
new file mode 100755
index 0000000..df2cf0a
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/JoinResponseMessage.java
@@ -0,0 +1,84 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messages;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.HighPriorityDistributionMessage;
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.distributed.internal.membership.NetView;
+import com.gemstone.gemfire.internal.DataSerializableFixedID;
+import com.gemstone.gemfire.internal.Version;
+
+public class JoinResponseMessage extends HighPriorityDistributionMessage {
+
+ private NetView currentView;
+ private String rejectionMessage;
+ private InternalDistributedMember memberID;
+
+ public JoinResponseMessage(InternalDistributedMember memberID, NetView view) {
+ this.currentView = view;
+ this.memberID = memberID;
+ setRecipient(memberID);
+ }
+
+ public JoinResponseMessage(String rejectionMessage) {
+ this.rejectionMessage = rejectionMessage;
+ }
+
+ public JoinResponseMessage() {
+ // no-arg constructor for serialization
+ }
+
+ public NetView getCurrentView() {
+ return currentView;
+ }
+
+ public InternalDistributedMember getMemberID() {
+ return memberID;
+ }
+
+ public String getRejectionMessage() {
+ return rejectionMessage;
+ }
+
+ @Override
+ public void process(DistributionManager dm) {
+ throw new IllegalStateException("JoinResponse is not intended to be executed");
+ }
+
+ @Override
+ public String toString() {
+ return getShortClassName() + "("+memberID + "; "
+ + (currentView==null? "" : currentView.toString())
+ + (rejectionMessage==null? "" : rejectionMessage)
+ + ")";
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+
+ @Override
+ public int getDSFID() {
+ return JOIN_RESPONSE;
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ DataSerializer.writeObject(currentView, out);
+ DataSerializer.writeObject(memberID, out);
+ DataSerializer.writeString(rejectionMessage, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ currentView = DataSerializer.readObject(in);
+ memberID = DataSerializer.readObject(in);
+ rejectionMessage = DataSerializer.readString(in);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java
new file mode 100755
index 0000000..632e35f
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java
@@ -0,0 +1,55 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messages;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.HighPriorityDistributionMessage;
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.internal.Version;
+
+public class LeaveRequestMessage extends HighPriorityDistributionMessage {
+ private InternalDistributedMember memberID;
+
+ public LeaveRequestMessage(InternalDistributedMember coord, InternalDistributedMember id) {
+ super();
+ setRecipient(coord);
+ this.memberID = id;
+ }
+
+ public LeaveRequestMessage() {
+ // no-arg constructor for serialization
+ }
+
+ @Override
+ public int getDSFID() {
+ return LEAVE_REQUEST_MESSAGE;
+ }
+
+ @Override
+ public void process(DistributionManager dm) {
+ throw new IllegalStateException("this message is not intended to execute in a thread pool");
+ }
+
+ public InternalDistributedMember getMemberID() {
+ return memberID;
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ DataSerializer.writeObject(memberID, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ memberID = DataSerializer.readObject(in);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/RemoveMemberMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/RemoveMemberMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/RemoveMemberMessage.java
new file mode 100755
index 0000000..e575ea0
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/RemoveMemberMessage.java
@@ -0,0 +1,70 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messages;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.HighPriorityDistributionMessage;
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.internal.Version;
+
+public class RemoveMemberMessage extends HighPriorityDistributionMessage {
+ private InternalDistributedMember memberID;
+ private String reason;
+
+
+ public RemoveMemberMessage(InternalDistributedMember recipient, InternalDistributedMember id, String reason) {
+ super();
+ setRecipient(recipient);
+ this.memberID = id;
+ this.reason = reason;
+ }
+
+ public RemoveMemberMessage() {
+ // no-arg constructor for serialization
+ }
+
+ @Override
+ public int getDSFID() {
+ return JOIN_REQUEST;
+ }
+
+ @Override
+ public void process(DistributionManager dm) {
+ throw new IllegalStateException("this message is not intended to execute in a thread pool");
+ }
+
+ public InternalDistributedMember getMemberID() {
+ return memberID;
+ }
+
+ public String getReason() {
+ return reason;
+ }
+
+ @Override
+ public String toString() {
+ return getShortClassName() + "(" + memberID
+ + "; reason=" + reason + ")";
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ DataSerializer.writeObject(memberID, out);
+ DataSerializer.writeString(reason, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ memberID = DataSerializer.readObject(in);
+ reason = DataSerializer.readString(in);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/ViewAckMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/ViewAckMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/ViewAckMessage.java
new file mode 100755
index 0000000..f52811b
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/ViewAckMessage.java
@@ -0,0 +1,74 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messages;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.DistributionMessage;
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.distributed.internal.membership.NetView;
+
+public class ViewAckMessage extends DistributionMessage {
+
+ int viewId;
+ NetView alternateView;
+
+ public ViewAckMessage(InternalDistributedMember recipient, int viewId) {
+ super();
+ setRecipient(recipient);
+ this.viewId = viewId;
+ }
+
+ public ViewAckMessage(InternalDistributedMember recipient, NetView alternateView) {
+ super();
+ setRecipient(recipient);
+ this.alternateView = alternateView;
+ }
+
+ public ViewAckMessage() {
+ // no-arg constructor for serialization
+ }
+
+ public int getViewId() {
+ return viewId;
+ }
+
+ @Override
+ public int getDSFID() {
+ // TODO Auto-generated method stub
+ return VIEW_ACK_MESSAGE;
+ }
+
+ @Override
+ public int getProcessorType() {
+ return 0;
+ }
+
+ @Override
+ protected void process(DistributionManager dm) {
+ throw new IllegalStateException("this message is not intended to execute in a thread pool");
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ super.toData(out);
+ out.writeInt(this.viewId);
+ DataSerializer.writeObject(this.alternateView, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ super.fromData(in);
+ this.viewId = in.readInt();
+ this.alternateView = DataSerializer.readObject(in);
+ }
+
+ @Override
+ public String toString() {
+ String s = getSender() == null? getRecipientsDescription() : ""+getSender();
+ return "ViewAckMessage("+s+"; "+this.viewId+"; altview="+this.alternateView+")";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/AddressManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/AddressManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/AddressManager.java
new file mode 100755
index 0000000..ff03bfd
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/AddressManager.java
@@ -0,0 +1,113 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messenger;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.List;
+
+import org.apache.logging.log4j.Logger;
+import org.jgroups.Address;
+import org.jgroups.Event;
+import org.jgroups.Message;
+import org.jgroups.protocols.PingData;
+import org.jgroups.protocols.TP;
+import org.jgroups.protocols.UDP;
+import org.jgroups.stack.IpAddress;
+import org.jgroups.stack.Protocol;
+import org.jgroups.util.Responses;
+
+import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
+
+/**
+ * JGroups will not send messages that have IpAddress destination addresses.
+ * Instead it requires a "logical address" and requests physical addresses
+ * from the Discovery protocol that is normally in a JGroups stack. We don't have
+ * one of these, so we need to maintain a mapping between logical and physical
+ * addresses.
+ *
+ * @author bschuchardt
+ *
+ */
+public class AddressManager extends Protocol {
+
+ private static final Logger logger = Services.getLogger();
+
+ private UDP udp;
+ private Method setPingData;
+ boolean warningLogged = false;
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object up(Event evt) {
+
+// logger.info("AddressManager.up: " + evt);
+
+ switch (evt.getType()) {
+
+ case Event.FIND_MBRS:
+ List<Address> missing = (List<Address>)evt.getArg();
+ logger.debug("AddressManager.FIND_MBRS processing {}", missing);
+ 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);
+ }
+
+ private IpAddress newIpAddress(Address jgaddr) {
+ JGAddress addr = (JGAddress)jgaddr;
+ return addr.asIpAddress();
+ }
+
+ /**
+ * update the logical->physical address cache in UDP, which doesn't
+ * seem to be updated by UDP when processing responses from FIND_MBRS
+ * @param pd
+ */
+ private void updateUDPCache(PingData pd) {
+ if (setPingData == null && !warningLogged) {
+ findPingDataMethod();
+ }
+ if (setPingData != null) {
+ Exception problem = null;
+ try {
+ setPingData.invoke(udp, new Object[]{pd});
+ } catch (InvocationTargetException e) {
+ problem = e;
+ } catch (IllegalAccessException e) {
+ problem = e;
+ }
+ if (problem != null && !warningLogged) {
+ log.warn("Unable to update JGroups address cache - this may affect performance", problem);
+ warningLogged = true;
+ }
+ }
+ }
+
+ /**
+ * find and initialize the method used to update UDP's address cache
+ */
+ private void findPingDataMethod() {
+ udp = (UDP)getProtocolStack().findProtocol("UDP");
+ try {
+ setPingData = TP.class.getDeclaredMethod("setPingData", new Class<?>[]{PingData.class});
+ setPingData.setAccessible(true);
+ } catch (NoSuchMethodException e) {
+ if (!warningLogged) {
+ log.warn("Unable to update JGroups address cache - this may affect performance", e);
+ warningLogged = true;
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/d2a942e8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGAddress.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGAddress.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGAddress.java
new file mode 100755
index 0000000..b34b8ed
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGAddress.java
@@ -0,0 +1,197 @@
+package com.gemstone.gemfire.distributed.internal.membership.gms.messenger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.net.Inet4Address;
+import java.net.Inet6Address;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.jgroups.Address;
+import org.jgroups.Global;
+import org.jgroups.stack.IpAddress;
+import org.jgroups.util.UUID;
+
+import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.distributed.internal.membership.gms.GMSMember;
+import com.gemstone.gemfire.internal.SocketCreator;
+
+/**
+ * This is a copy of JGroups 3.6.4 IpAddress (Apache 2.0 License)
+ * that is repurposed to be a Logical address so that we can
+ * quickly pull a physical address out of the logical address set
+ * in a message by JGroupsMessenger.
+ */
+
+public class JGAddress extends UUID {
+ private static final long serialVersionUID=-1818672332115113291L;
+ private InetAddress ip_addr;
+ private int port;
+ private int vmViewId;
+
+
+ // Used only by Externalization
+ public JGAddress() {
+ }
+
+ public JGAddress(InternalDistributedMember idm) {
+ super();
+ GMSMember mbr = (GMSMember)idm.getNetMember();
+ this.mostSigBits = mbr.getUuidMSBs();
+ this.leastSigBits = mbr.getUuidLSBs();
+ this.ip_addr = idm.getInetAddress();
+ this.port = idm.getPort();
+ this.vmViewId = idm.getVmViewId();
+ }
+
+
+ public JGAddress(GMSMember mbr) {
+ super();
+ this.mostSigBits = mbr.getUuidMSBs();
+ this.leastSigBits = mbr.getUuidLSBs();
+ this.ip_addr = mbr.getInetAddress();
+ this.port = mbr.getPort();
+ this.vmViewId = mbr.getVmViewId();
+ }
+
+
+ public JGAddress(UUID uuid, IpAddress ipaddr) {
+ super(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
+ ip_addr=ipaddr.getIpAddress(); port=ipaddr.getPort(); vmViewId=-1;
+ }
+
+
+ private void setAddressToLocalHost() {
+ try {
+ ip_addr=InetAddress.getLocalHost(); // get first NIC found (on multi-homed systems)
+ }
+ catch(Exception e) {
+ ip_addr=null;
+ }
+ if(ip_addr == null) {
+ try {
+ ip_addr=InetAddress.getByName(null);
+ }
+ catch(UnknownHostException e) {
+ }
+ }
+ }
+
+ public final InetAddress getInetAddress() {return ip_addr;}
+ public final int getPort() {return port;}
+
+ public final int getVmViewId() {
+ return this.vmViewId;
+ }
+
+ protected final void setVmViewId(int id) {
+ this.vmViewId = id;
+ }
+
+
+ public String toString() {
+ StringBuilder sb=new StringBuilder();
+
+ if(ip_addr == null)
+ sb.append("<null>");
+ else {
+ if (SocketCreator.resolve_dns) {
+ sb.append(SocketCreator.getHostName(ip_addr));
+ } else {
+ sb.append(ip_addr.getHostAddress());
+ }
+ }
+ if (vmViewId >= 0) {
+ sb.append("<v").append(vmViewId).append('>');
+ }
+ sb.append(":").append(port);
+ return sb.toString();
+ }
+
+
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ try {
+ readFrom(in);
+ }
+ catch(Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ public void writeExternal(ObjectOutput out) throws IOException {
+ try {
+ writeTo(out);
+ }
+ catch(Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ public void writeTo(DataOutput out) throws Exception {
+ if(ip_addr != null) {
+ byte[] address=ip_addr.getAddress(); // 4 bytes (IPv4) or 16 bytes (IPv6)
+ out.writeByte(address.length); // 1 byte
+ out.write(address, 0, address.length);
+ if(ip_addr instanceof Inet6Address)
+ out.writeInt(((Inet6Address)ip_addr).getScopeId());
+ }
+ else {
+ out.writeByte(0);
+ }
+ out.writeShort(port);
+ out.writeInt(vmViewId);
+ out.writeLong(mostSigBits);
+ out.writeLong(leastSigBits);
+ }
+
+ public void readFrom(DataInput in) throws Exception {
+ int len=in.readByte();
+ if(len > 0 && (len != Global.IPV4_SIZE && len != Global.IPV6_SIZE))
+ throw new IOException("length has to be " + Global.IPV4_SIZE + " or " + Global.IPV6_SIZE + " bytes (was " +
+ len + " bytes)");
+ byte[] a = new byte[len]; // 4 bytes (IPv4) or 16 bytes (IPv6)
+ in.readFully(a);
+ if(len == Global.IPV6_SIZE) {
+ int scope_id=in.readInt();
+ this.ip_addr=Inet6Address.getByAddress(null, a, scope_id);
+ }
+ else {
+ this.ip_addr=InetAddress.getByAddress(a);
+ }
+
+ // changed from readShort(): we need the full 65535, with a short we'd only get up to 32K !
+ port=in.readUnsignedShort();
+ vmViewId = in.readInt();
+ mostSigBits = in.readLong();
+ leastSigBits = in.readLong();
+ }
+
+ public int size() {
+ // length (1 bytes) + 4 bytes for port
+ int tmp_size=Global.BYTE_SIZE+ Global.SHORT_SIZE +Global.SHORT_SIZE
+ + (2*Global.LONG_SIZE);
+ if(ip_addr != null) {
+ // 4 bytes for IPv4, 20 for IPv6 (16 + 4 for scope-id)
+ tmp_size+=(ip_addr instanceof Inet4Address)? 4 : 20;
+ }
+ return tmp_size;
+ }
+
+ public JGAddress copy() {
+ JGAddress result = new JGAddress();
+ result.mostSigBits = mostSigBits;
+ result.leastSigBits = leastSigBits;
+ result.ip_addr = ip_addr;
+ result.port = port;
+ result.vmViewId = vmViewId;
+ return result;
+ }
+
+ public IpAddress asIpAddress() {
+ return new IpAddress(ip_addr, port);
+ }
+
+}