geode-core/src/main/java/org/apache/geode/management/internal/JmxManagerAdvisor.java - geode - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
  * agreements. See the NOTICE file distributed with this work for additional information regarding
  * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance with the License. You may obtain a
  * copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License
  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  */
 package org.apache.geode.management.internal;

 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.CancelException;
 import org.apache.geode.DataSerializer;
 import org.apache.geode.SystemFailure;
 import org.apache.geode.distributed.internal.ClusterDistributionManager;
 import org.apache.geode.distributed.internal.DistributionAdvisee;
 import org.apache.geode.distributed.internal.DistributionAdvisor;
 import org.apache.geode.distributed.internal.DistributionManager;
 import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.serialization.DeserializationContext;
 import org.apache.geode.internal.serialization.SerializationContext;
 import org.apache.geode.logging.internal.log4j.api.LogService;

 /**
  * @since GemFire 7.0
  */
 public class JmxManagerAdvisor extends DistributionAdvisor {

   private static final Logger logger = LogService.getLogger();

   private JmxManagerAdvisor(DistributionAdvisee advisee) {
     super(advisee);
     JmxManagerProfile p =
         new JmxManagerProfile(getDistributionManager().getId(), incrementAndGetVersion());
     advisee.fillInProfile(p);
     ((JmxManagerAdvisee) advisee).initProfile(p);
   }

   public static JmxManagerAdvisor createJmxManagerAdvisor(DistributionAdvisee advisee) {
     JmxManagerAdvisor advisor = new JmxManagerAdvisor(advisee);
     advisor.initialize();
     return advisor;
   }

   @Override
   public String toString() {
     return new StringBuilder().append("JmxManagerAdvisor for " + getAdvisee()).toString();
   }

   public void broadcastChange() {
     try {
       Set<InternalDistributedMember> recips = adviseGeneric(); // for now just tell everyone
       JmxManagerProfile p =
           new JmxManagerProfile(getDistributionManager().getId(), incrementAndGetVersion());
       getAdvisee().fillInProfile(p);
       JmxManagerProfileMessage.send(getAdvisee().getSystem().getDistributionManager(), recips, p);
     } catch (CancelException ignore) {
     }
   }

   @SuppressWarnings("unchecked")
   public List<JmxManagerProfile> adviseAlreadyManaging() {
     return fetchProfiles(new Filter() {
       @Override
       public boolean include(Profile profile) {
         assert profile instanceof JmxManagerProfile;
         JmxManagerProfile jmxProfile = (JmxManagerProfile) profile;
         return jmxProfile.isJmxManagerRunning();
       }
     });
   }

   @SuppressWarnings("unchecked")
   public List<JmxManagerProfile> adviseWillingToManage() {
     return fetchProfiles(new Filter() {
       @Override
       public boolean include(Profile profile) {
         assert profile instanceof JmxManagerProfile;
         JmxManagerProfile jmxProfile = (JmxManagerProfile) profile;
         return jmxProfile.isJmxManager();
       }
     });
   }

   @Override
   protected Profile instantiateProfile(InternalDistributedMember memberId, int version) {
     return new JmxManagerProfile(memberId, version);
   }

   /**
    * Overridden to also include our profile. If our profile is included it will always be first.
    */
   @Override
   protected List/* <Profile> */ fetchProfiles(Filter f) {
     initializationGate();
     List result = null;
     {
       JmxManagerAdvisee advisee = (JmxManagerAdvisee) getAdvisee();
       Profile myp = advisee.getMyMostRecentProfile();
       if (f == null || f.include(myp)) {
         if (result == null) {
           result = new ArrayList();
         }
         result.add(myp);
       }
     }
     Profile[] locProfiles = this.profiles; // grab current profiles
     for (int i = 0; i < locProfiles.length; i++) {
       Profile profile = locProfiles[i];
       if (f == null || f.include(profile)) {
         if (result == null) {
           result = new ArrayList(locProfiles.length);
         }
         result.add(profile);
       }
     }
     if (result == null) {
       result = Collections.EMPTY_LIST;
     } else {
       result = Collections.unmodifiableList(result);
     }
     return result;
   }

   /**
    * Message used to push event updates to remote VMs
    */
   public static class JmxManagerProfileMessage extends HighPriorityDistributionMessage {
     private volatile JmxManagerProfile profile;
     private volatile int processorId;

     /**
      * Default constructor used for de-serialization (used during receipt)
      */
     public JmxManagerProfileMessage() {}

     @Override
     public boolean sendViaUDP() {
       return true;
     }

     /**
      * Constructor used to send
      *
      */
     private JmxManagerProfileMessage(final Set<InternalDistributedMember> recips,
         final JmxManagerProfile p) {
       setRecipients(recips);
       this.processorId = 0;
       this.profile = p;
     }

     @Override
     protected void process(ClusterDistributionManager dm) {
       Throwable thr = null;
       JmxManagerProfile p = null;
       try {
         final InternalCache cache = dm.getCache();
         if (cache != null && !cache.isClosed()) {
           final JmxManagerAdvisor adv = cache.getJmxManagerAdvisor();
           p = this.profile;
           if (p != null) {
             adv.putProfile(p);
           }
         } else {
           if (logger.isDebugEnabled()) {
             logger.debug("No cache {}", this);
           }
         }
       } catch (CancelException e) {
         if (logger.isDebugEnabled()) {
           logger.debug("Cache closed, {}", this);
         }
       } catch (VirtualMachineError err) {
         SystemFailure.initiateFailure(err);
         // If this ever returns, rethrow the error. We're poisoned
         // now, so don't let this thread continue.
         throw err;
       } catch (Throwable t) {
         // Whenever you catch Error or Throwable, you must also
         // catch VirtualMachineError (see above). However, there is
         // _still_ a possibility that you are dealing with a cascading
         // error condition, so you also need to check to see if the JVM
         // is still usable:
         SystemFailure.checkFailure();
         thr = t;
       } finally {
         if (thr != null) {
           dm.getCancelCriterion().checkCancelInProgress(null);
           logger.info(String.format("This member caught exception processing profile %s %s",
               p, toString()), thr);
         }
       }
     }

     @Override
     public int getDSFID() {
       return JMX_MANAGER_PROFILE_MESSAGE;
     }

     @Override
     public void fromData(DataInput in,
         DeserializationContext context) throws IOException, ClassNotFoundException {
       super.fromData(in, context);
       this.processorId = in.readInt();
       this.profile = (JmxManagerProfile) DataSerializer.readObject(in);
     }

     @Override
     public void toData(DataOutput out,
         SerializationContext context) throws IOException {
       super.toData(out, context);
       out.writeInt(this.processorId);
       DataSerializer.writeObject(this.profile, out);
     }

     /**
      * Send profile to the provided members
      *
      * @param recips The recipients of the message
      */
     public static void send(final DistributionManager dm, Set<InternalDistributedMember> recips,
         JmxManagerProfile profile) {
       JmxManagerProfileMessage r = new JmxManagerProfileMessage(recips, profile);
       dm.putOutgoing(r);
     }

     @Override
     public String getShortClassName() {
       return "JmxManagerProfileMessage";
     }

     @Override
     public String toString() {
       StringBuilder sb = new StringBuilder();
       sb.append(getShortClassName()).append(" (processorId=").append(this.processorId)
           .append("; profile=").append(this.profile);
       sb.append(")");
       return sb.toString();
     }
   }

   public static class JmxManagerProfile extends Profile {

     private boolean jmxManager;
     private String host;
     private int port;
     private boolean ssl;
     private boolean started;

     // Constructor for de-serialization
     public JmxManagerProfile() {}

     public boolean isJmxManager() {
       return this.jmxManager;
     }

     public boolean isJmxManagerRunning() {
       return this.started;
     }

     public void setInfo(boolean jmxManager2, String host2, int port2, boolean ssl2,
         boolean started2) {
       this.jmxManager = jmxManager2;
       this.host = host2;
       this.port = port2;
       this.ssl = ssl2;
       this.started = started2;
     }

     public String getHost() {
       return this.host;
     }

     public int getPort() {
       return this.port;
     }

     public boolean getSsl() {
       return this.ssl;
     }

     // Constructor for sending purposes
     public JmxManagerProfile(InternalDistributedMember memberId, int version) {
       super(memberId, version);
     }

     @Override
     public StringBuilder getToStringHeader() {
       return new StringBuilder("JmxManagerAdvisor.JmxManagerProfile");
     }

     @Override
     public void fillInToString(StringBuilder sb) {
       super.fillInToString(sb);
       synchronized (this) {
         if (this.jmxManager) {
           sb.append("; jmxManager");
         }
         sb.append("; host=").append(this.host).append("; port=").append(this.port);
         if (this.ssl) {
           sb.append("; ssl");
         }
         if (this.started) {
           sb.append("; started");
         }
       }
     }

     @Override
     public void processIncoming(ClusterDistributionManager dm, String adviseePath,
         boolean removeProfile, boolean exchangeProfiles, final List<Profile> replyProfiles) {
       final InternalCache cache = dm.getCache();
       if (cache != null && !cache.isClosed()) {
         handleDistributionAdvisee(cache.getJmxManagerAdvisor().getAdvisee(), removeProfile,
             exchangeProfiles, replyProfiles);
       }
     }

     @Override
     public void fromData(DataInput in,
         DeserializationContext context) throws IOException, ClassNotFoundException {
       super.fromData(in, context);
       this.jmxManager = DataSerializer.readPrimitiveBoolean(in);
       this.host = DataSerializer.readString(in);
       this.port = DataSerializer.readPrimitiveInt(in);
       this.ssl = DataSerializer.readPrimitiveBoolean(in);
       this.started = DataSerializer.readPrimitiveBoolean(in);
     }

     @Override
     public void toData(DataOutput out,
         SerializationContext context) throws IOException {
       boolean tmpJmxManager;
       String tmpHost;
       int tmpPort;
       boolean tmpSsl;
       boolean tmpStarted;
       synchronized (this) {
         tmpJmxManager = this.jmxManager;
         tmpHost = this.host;
         tmpPort = this.port;
         tmpSsl = this.ssl;
         tmpStarted = this.started;
       }
       super.toData(out, context);
       DataSerializer.writePrimitiveBoolean(tmpJmxManager, out);
       DataSerializer.writeString(tmpHost, out);
       DataSerializer.writePrimitiveInt(tmpPort, out);
       DataSerializer.writePrimitiveBoolean(tmpSsl, out);
       DataSerializer.writePrimitiveBoolean(tmpStarted, out);
     }

     @Override
     public int getDSFID() {
       return JMX_MANAGER_PROFILE;
     }
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more contributor license
	* agreements. See the NOTICE file distributed with this work for additional information regarding
	* copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance with the License. You may obtain a
	* copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software distributed under the License
	* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
	* or implied. See the License for the specific language governing permissions and limitations under
	* the License.
	*/
	package org.apache.geode.management.internal;

	import java.io.DataInput;
	import java.io.DataOutput;
	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.List;
	import java.util.Set;

	import org.apache.logging.log4j.Logger;

	import org.apache.geode.CancelException;
	import org.apache.geode.DataSerializer;
	import org.apache.geode.SystemFailure;
	import org.apache.geode.distributed.internal.ClusterDistributionManager;
	import org.apache.geode.distributed.internal.DistributionAdvisee;
	import org.apache.geode.distributed.internal.DistributionAdvisor;
	import org.apache.geode.distributed.internal.DistributionManager;
	import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
	import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
	import org.apache.geode.internal.cache.InternalCache;
	import org.apache.geode.internal.serialization.DeserializationContext;
	import org.apache.geode.internal.serialization.SerializationContext;
	import org.apache.geode.logging.internal.log4j.api.LogService;

	/**
	* @since GemFire 7.0
	*/
	public class JmxManagerAdvisor extends DistributionAdvisor {

	private static final Logger logger = LogService.getLogger();

	private JmxManagerAdvisor(DistributionAdvisee advisee) {
	super(advisee);
	JmxManagerProfile p =
	new JmxManagerProfile(getDistributionManager().getId(), incrementAndGetVersion());
	advisee.fillInProfile(p);
	((JmxManagerAdvisee) advisee).initProfile(p);
	}

	public static JmxManagerAdvisor createJmxManagerAdvisor(DistributionAdvisee advisee) {
	JmxManagerAdvisor advisor = new JmxManagerAdvisor(advisee);
	advisor.initialize();
	return advisor;
	}

	@Override
	public String toString() {
	return new StringBuilder().append("JmxManagerAdvisor for " + getAdvisee()).toString();
	}

	public void broadcastChange() {
	try {
	Set<InternalDistributedMember> recips = adviseGeneric(); // for now just tell everyone
	JmxManagerProfile p =
	new JmxManagerProfile(getDistributionManager().getId(), incrementAndGetVersion());
	getAdvisee().fillInProfile(p);
	JmxManagerProfileMessage.send(getAdvisee().getSystem().getDistributionManager(), recips, p);
	} catch (CancelException ignore) {
	}
	}

	@SuppressWarnings("unchecked")
	public List<JmxManagerProfile> adviseAlreadyManaging() {
	return fetchProfiles(new Filter() {
	@Override
	public boolean include(Profile profile) {
	assert profile instanceof JmxManagerProfile;
	JmxManagerProfile jmxProfile = (JmxManagerProfile) profile;
	return jmxProfile.isJmxManagerRunning();
	}
	});
	}

	@SuppressWarnings("unchecked")
	public List<JmxManagerProfile> adviseWillingToManage() {
	return fetchProfiles(new Filter() {
	@Override
	public boolean include(Profile profile) {
	assert profile instanceof JmxManagerProfile;
	JmxManagerProfile jmxProfile = (JmxManagerProfile) profile;
	return jmxProfile.isJmxManager();
	}
	});
	}

	@Override
	protected Profile instantiateProfile(InternalDistributedMember memberId, int version) {
	return new JmxManagerProfile(memberId, version);
	}

	/**
	* Overridden to also include our profile. If our profile is included it will always be first.
	*/
	@Override
	protected List/* <Profile> */ fetchProfiles(Filter f) {
	initializationGate();
	List result = null;
	{
	JmxManagerAdvisee advisee = (JmxManagerAdvisee) getAdvisee();
	Profile myp = advisee.getMyMostRecentProfile();
	if (f == null \|\| f.include(myp)) {
	if (result == null) {
	result = new ArrayList();
	}
	result.add(myp);
	}
	}
	Profile[] locProfiles = this.profiles; // grab current profiles
	for (int i = 0; i < locProfiles.length; i++) {
	Profile profile = locProfiles[i];
	if (f == null \|\| f.include(profile)) {
	if (result == null) {
	result = new ArrayList(locProfiles.length);
	}
	result.add(profile);
	}
	}
	if (result == null) {
	result = Collections.EMPTY_LIST;
	} else {
	result = Collections.unmodifiableList(result);
	}
	return result;
	}

	/**
	* Message used to push event updates to remote VMs
	*/
	public static class JmxManagerProfileMessage extends HighPriorityDistributionMessage {
	private volatile JmxManagerProfile profile;
	private volatile int processorId;

	/**
	* Default constructor used for de-serialization (used during receipt)
	*/
	public JmxManagerProfileMessage() {}

	@Override
	public boolean sendViaUDP() {
	return true;
	}

	/**
	* Constructor used to send
	*
	*/
	private JmxManagerProfileMessage(final Set<InternalDistributedMember> recips,
	final JmxManagerProfile p) {
	setRecipients(recips);
	this.processorId = 0;
	this.profile = p;
	}

	@Override
	protected void process(ClusterDistributionManager dm) {
	Throwable thr = null;
	JmxManagerProfile p = null;
	try {
	final InternalCache cache = dm.getCache();
	if (cache != null && !cache.isClosed()) {
	final JmxManagerAdvisor adv = cache.getJmxManagerAdvisor();
	p = this.profile;
	if (p != null) {
	adv.putProfile(p);
	}
	} else {
	if (logger.isDebugEnabled()) {
	logger.debug("No cache {}", this);
	}
	}
	} catch (CancelException e) {
	if (logger.isDebugEnabled()) {
	logger.debug("Cache closed, {}", this);
	}
	} catch (VirtualMachineError err) {
	SystemFailure.initiateFailure(err);
	// If this ever returns, rethrow the error. We're poisoned
	// now, so don't let this thread continue.
	throw err;
	} catch (Throwable t) {
	// Whenever you catch Error or Throwable, you must also
	// catch VirtualMachineError (see above). However, there is
	// _still_ a possibility that you are dealing with a cascading
	// error condition, so you also need to check to see if the JVM
	// is still usable:
	SystemFailure.checkFailure();
	thr = t;
	} finally {
	if (thr != null) {
	dm.getCancelCriterion().checkCancelInProgress(null);
	logger.info(String.format("This member caught exception processing profile %s %s",
	p, toString()), thr);
	}
	}
	}

	@Override
	public int getDSFID() {
	return JMX_MANAGER_PROFILE_MESSAGE;
	}

	@Override
	public void fromData(DataInput in,
	DeserializationContext context) throws IOException, ClassNotFoundException {
	super.fromData(in, context);
	this.processorId = in.readInt();
	this.profile = (JmxManagerProfile) DataSerializer.readObject(in);
	}

	@Override
	public void toData(DataOutput out,
	SerializationContext context) throws IOException {
	super.toData(out, context);
	out.writeInt(this.processorId);
	DataSerializer.writeObject(this.profile, out);
	}

	/**
	* Send profile to the provided members
	*
	* @param recips The recipients of the message
	*/
	public static void send(final DistributionManager dm, Set<InternalDistributedMember> recips,
	JmxManagerProfile profile) {
	JmxManagerProfileMessage r = new JmxManagerProfileMessage(recips, profile);
	dm.putOutgoing(r);
	}

	@Override
	public String getShortClassName() {
	return "JmxManagerProfileMessage";
	}

	@Override
	public String toString() {
	StringBuilder sb = new StringBuilder();
	sb.append(getShortClassName()).append(" (processorId=").append(this.processorId)
	.append("; profile=").append(this.profile);
	sb.append(")");
	return sb.toString();
	}
	}

	public static class JmxManagerProfile extends Profile {

	private boolean jmxManager;
	private String host;
	private int port;
	private boolean ssl;
	private boolean started;

	// Constructor for de-serialization
	public JmxManagerProfile() {}

	public boolean isJmxManager() {
	return this.jmxManager;
	}

	public boolean isJmxManagerRunning() {
	return this.started;
	}

	public void setInfo(boolean jmxManager2, String host2, int port2, boolean ssl2,
	boolean started2) {
	this.jmxManager = jmxManager2;
	this.host = host2;
	this.port = port2;
	this.ssl = ssl2;
	this.started = started2;
	}

	public String getHost() {
	return this.host;
	}

	public int getPort() {
	return this.port;
	}

	public boolean getSsl() {
	return this.ssl;
	}

	// Constructor for sending purposes
	public JmxManagerProfile(InternalDistributedMember memberId, int version) {
	super(memberId, version);
	}

	@Override
	public StringBuilder getToStringHeader() {
	return new StringBuilder("JmxManagerAdvisor.JmxManagerProfile");
	}

	@Override
	public void fillInToString(StringBuilder sb) {
	super.fillInToString(sb);
	synchronized (this) {
	if (this.jmxManager) {
	sb.append("; jmxManager");
	}
	sb.append("; host=").append(this.host).append("; port=").append(this.port);
	if (this.ssl) {
	sb.append("; ssl");
	}
	if (this.started) {
	sb.append("; started");
	}
	}
	}

	@Override
	public void processIncoming(ClusterDistributionManager dm, String adviseePath,
	boolean removeProfile, boolean exchangeProfiles, final List<Profile> replyProfiles) {
	final InternalCache cache = dm.getCache();
	if (cache != null && !cache.isClosed()) {
	handleDistributionAdvisee(cache.getJmxManagerAdvisor().getAdvisee(), removeProfile,
	exchangeProfiles, replyProfiles);
	}
	}

	@Override
	public void fromData(DataInput in,
	DeserializationContext context) throws IOException, ClassNotFoundException {
	super.fromData(in, context);
	this.jmxManager = DataSerializer.readPrimitiveBoolean(in);
	this.host = DataSerializer.readString(in);
	this.port = DataSerializer.readPrimitiveInt(in);
	this.ssl = DataSerializer.readPrimitiveBoolean(in);
	this.started = DataSerializer.readPrimitiveBoolean(in);
	}

	@Override
	public void toData(DataOutput out,
	SerializationContext context) throws IOException {
	boolean tmpJmxManager;
	String tmpHost;
	int tmpPort;
	boolean tmpSsl;
	boolean tmpStarted;
	synchronized (this) {
	tmpJmxManager = this.jmxManager;
	tmpHost = this.host;
	tmpPort = this.port;
	tmpSsl = this.ssl;
	tmpStarted = this.started;
	}
	super.toData(out, context);
	DataSerializer.writePrimitiveBoolean(tmpJmxManager, out);
	DataSerializer.writeString(tmpHost, out);
	DataSerializer.writePrimitiveInt(tmpPort, out);
	DataSerializer.writePrimitiveBoolean(tmpSsl, out);
	DataSerializer.writePrimitiveBoolean(tmpStarted, out);
	}

	@Override
	public int getDSFID() {
	return JMX_MANAGER_PROFILE;
	}
	}

	}