package org.jboss.invocation.jrmp.interfaces;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.rmi.MarshalledObject;
import java.rmi.ServerException;
import java.util.ArrayList;
import java.util.WeakHashMap;
import javax.transaction.TransactionRolledbackException;
import org.jboss.ha.framework.interfaces.ClusteringTargetsRepository;
import org.jboss.ha.framework.interfaces.FamilyClusterInfo;
import org.jboss.ha.framework.interfaces.GenericClusteringException;
import org.jboss.ha.framework.interfaces.HARMIResponse;
import org.jboss.ha.framework.interfaces.LoadBalancePolicy;
import org.jboss.invocation.Invocation;
import org.jboss.invocation.Invoker;
import org.jboss.invocation.InvokerProxyHA;
import org.jboss.invocation.MarshalledInvocation;
import org.jboss.invocation.PayloadKey;
import org.jboss.invocation.ServiceUnavailableException;
import org.jboss.logging.Logger;
public class JRMPInvokerProxyHA
extends JRMPInvokerProxy
implements InvokerProxyHA, Externalizable
{
private static final long serialVersionUID = -967671822225981666L;
private static final Logger log = Logger.getLogger(JRMPInvokerProxyHA.class);
public static final WeakHashMap txFailoverAuthorizations = new WeakHashMap();
protected LoadBalancePolicy loadBalancePolicy;
protected String proxyFamilyName = null;
FamilyClusterInfo familyClusterInfo = null;
protected transient boolean trace = false;
public JRMPInvokerProxyHA() {}
public JRMPInvokerProxyHA(ArrayList targets, LoadBalancePolicy policy,
String proxyFamilyName, long viewId)
{
this.familyClusterInfo = ClusteringTargetsRepository.initTarget (proxyFamilyName, targets, viewId);
this.loadBalancePolicy = policy;
this.proxyFamilyName = proxyFamilyName;
this.trace = log.isTraceEnabled();
if( trace )
log.trace("Init, cluterInfo: "+familyClusterInfo+", policy="+loadBalancePolicy);
}
public void updateClusterInfo (ArrayList targets, long viewId)
{
if (familyClusterInfo != null)
this.familyClusterInfo.updateClusterInfo (targets, viewId);
}
public Object getRemoteTarget()
{
return getRemoteTarget(null);
}
public Object getRemoteTarget(Invocation invocationBasedRouting)
{
return loadBalancePolicy.chooseTarget(this.familyClusterInfo, invocationBasedRouting);
}
public void remoteTargetHasFailed(Object target)
{
removeDeadTarget(target);
}
protected void removeDeadTarget(Object target)
{
if (this.familyClusterInfo != null)
this.familyClusterInfo.removeDeadTarget (target);
}
protected int totalNumberOfTargets ()
{
if (this.familyClusterInfo != null)
return this.familyClusterInfo.getTargets ().size ();
else
return 0;
}
protected void resetView ()
{
this.familyClusterInfo.resetView ();
}
public boolean txContextAllowsFailover (Invocation invocation)
{
javax.transaction.Transaction tx = invocation.getTransaction();
if (tx != null)
{
synchronized (tx)
{
return ! txFailoverAuthorizations.containsKey (tx);
}
}
else
{
return true;
}
}
public void invocationHasReachedAServer (Invocation invocation)
{
javax.transaction.Transaction tx = invocation.getTransaction();
if (tx != null)
{
synchronized (tx)
{
txFailoverAuthorizations.put (tx, null);
}
}
}
public Object invoke(Invocation invocation)
throws Exception
{
int failoverCounter = 0;
invocation.setValue ("FAILOVER_COUNTER", new Integer(failoverCounter), PayloadKey.AS_IS);
MarshalledInvocation mi = new MarshalledInvocation(invocation);
mi.setTransactionPropagationContext(getTransactionPropagationContext());
mi.setValue("CLUSTER_VIEW_ID", new Long(this.familyClusterInfo.getCurrentViewId ()));
Invoker target = (Invoker)getRemoteTarget(invocation);
boolean failoverAuthorized = true;
Exception lastException = null;
while (target != null && failoverAuthorized)
{
boolean definitivlyRemoveNodeOnFailure = true;
try
{
if( trace )
log.trace("Invoking on target="+target);
Object rtnObj = target.invoke(mi);
HARMIResponse rsp = null;
if (rtnObj instanceof MarshalledObject)
{
rsp = (HARMIResponse)((MarshalledObject)rtnObj).get();
}
else
{
rsp = (HARMIResponse)rtnObj;
}
if (rsp.newReplicants != null)
{
if( trace )
{
log.trace("newReplicants: "+rsp.newReplicants);
}
updateClusterInfo (rsp.newReplicants, rsp.currentViewId);
}
invocationHasReachedAServer (invocation);
return rsp.response;
}
catch (java.net.ConnectException e)
{
lastException = e;
}
catch (java.net.UnknownHostException e)
{
lastException = e;
}
catch (java.rmi.ConnectException e)
{
lastException = e;
}
catch (java.rmi.ConnectIOException e)
{
lastException = e;
}
catch (java.rmi.NoSuchObjectException e)
{
lastException = e;
}
catch (java.rmi.UnmarshalException e)
{
lastException = e;
}
catch (java.rmi.UnknownHostException e)
{
lastException = e;
}
catch (GenericClusteringException e)
{
lastException = e;
if (e.getCompletionStatus () == GenericClusteringException.COMPLETED_NO)
{
if (totalNumberOfTargets() >= failoverCounter)
{
if (!e.isDefinitive ())
definitivlyRemoveNodeOnFailure = false;
}
}
else
{
invocationHasReachedAServer (invocation);
throw new ServerException("Clustering error", e);
}
}
catch (ServerException e)
{
invocationHasReachedAServer (invocation);
if (e.detail instanceof TransactionRolledbackException)
{
throw (TransactionRolledbackException) e.detail;
}
throw e;
}
catch (Exception e)
{
lastException = e;
invocationHasReachedAServer (invocation);
throw e;
}
if( trace )
log.trace("Invoke failed, target="+target, lastException);
remoteTargetHasFailed(target);
if (!definitivlyRemoveNodeOnFailure)
{
resetView ();
}
failoverAuthorized = txContextAllowsFailover (invocation);
target = (Invoker)getRemoteTarget(invocation);
failoverCounter++;
mi.setValue ("FAILOVER_COUNTER", new Integer(failoverCounter), PayloadKey.AS_IS);
}
String msg = "Service unavailable.";
if (failoverAuthorized == false)
{
msg = "Service unavailable (failover not possible inside a user transaction).";
}
throw new ServiceUnavailableException(msg, lastException);
}
public void writeExternal(final ObjectOutput out)
throws IOException
{
ArrayList targets = this.familyClusterInfo.getTargets();
long vid = this.familyClusterInfo.getCurrentViewId ();
targets.trimToSize();
out.writeObject(targets);
out.writeObject(this.loadBalancePolicy);
out.writeObject (this.proxyFamilyName);
out.writeLong (vid);
}
public void readExternal(final ObjectInput in)
throws IOException, ClassNotFoundException
{
ArrayList targets = (ArrayList)in.readObject();
this.loadBalancePolicy = (LoadBalancePolicy)in.readObject();
this.proxyFamilyName = (String)in.readObject();
long vid = in.readLong ();
this.familyClusterInfo = ClusteringTargetsRepository.initTarget (this.proxyFamilyName, targets, vid);
this.trace = log.isTraceEnabled();
if( trace )
log.trace("Init, clusterInfo: "+familyClusterInfo+", policy="+loadBalancePolicy);
}
}