text:
package org.apache.geode.cache.client.internal;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.GemFireConfigException;
import org.apache.geode.GemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.cache.InterestResultPolicy;
import org.apache.geode.cache.NoSubscriptionServersAvailableException;
import org.apache.geode.cache.client.ServerConnectivityException;
import org.apache.geode.cache.client.internal.PoolImpl.PoolTask;
import org.apache.geode.cache.client.internal.RegisterInterestTracker.RegionInterestEntry;
import org.apache.geode.cache.client.internal.ServerDenyList.DenyListListener;
import org.apache.geode.cache.client.internal.ServerDenyList.DenyListListenerAdapter;
import org.apache.geode.cache.client.internal.ServerDenyList.FailureTracker;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.cq.ClientCQ;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.ClientServerObserver;
import org.apache.geode.internal.cache.ClientServerObserverHolder;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ServerQueueStatus;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.LoggingExecutors;
import org.apache.geode.security.GemFireSecurityException;

/**
 * Manages Client Queues. Responsible for creating callback connections and satisfying redundancy
 * requirements.
 *
 * @since GemFire 5.7
 */
public class QueueManagerImpl implements QueueManager {
  private static final Logger logger = LogService.getLogger();

  private static final Comparator QSIZE_COMPARATOR = new QSizeComparator();

  protected final long redundancyRetryInterval;
  private final EndpointManager endpointManager;
  private final EndpointManager.EndpointListenerAdapter endpointListener;
  private final ConnectionSource source;
  private final int redundancyLevel;
  protected final ConnectionFactory factory;
  private final InternalLogWriter securityLogger;
  private final ClientProxyMembershipID proxyId;
  protected final InternalPool pool;
  private final QueueStateImpl state;
  private boolean printPrimaryNotFoundError;
  private boolean printRedundancyNotSatisfiedError;
  private boolean printRecoveringPrimary;
  private boolean printRecoveringRedundant;
  protected final ServerDenyList denyList;
  // Lock which guards updates to queueConnections.
  // Also threads calling getAllConnections will wait on this
  // lock until there is a primary.
  protected final Object lock = new Object();

  protected final CountDownLatch initializedLatch = new CountDownLatch(1);

  private ScheduledExecutorService recoveryThread;
  private volatile boolean sentClientReady;

  // queueConnections in maintained by using copy-on-write
  protected volatile ConnectionList queueConnections = new ConnectionList();
  protected volatile RedundancySatisfierTask redundancySatisfierTask = null;
  private volatile boolean shuttingDown;

  public QueueManagerImpl(InternalPool pool, EndpointManager endpointManager,
      ConnectionSource source, ConnectionFactory factory, int queueRedundancyLevel,
      long redundancyRetryInterval, InternalLogWriter securityLogger,
      ClientProxyMembershipID proxyId) {
    this.printPrimaryNotFoundError = true;
    this.printRedundancyNotSatisfiedError = true;
    this.printRecoveringRedundant = true;
    this.printRecoveringPrimary = true;
    this.pool = pool;
    this.endpointManager = endpointManager;
    this.source = source;
    this.factory = factory;
    this.redundancyLevel = queueRedundancyLevel;
    this.securityLogger = securityLogger;
    this.proxyId = proxyId;
    this.redundancyRetryInterval = redundancyRetryInterval;
    denyList = new ServerDenyList(redundancyRetryInterval);

    this.endpointListener = new EndpointManager.EndpointListenerAdapter() {
      @Override
      public void endpointCrashed(Endpoint endpoint) {
        QueueManagerImpl.this.endpointCrashed(endpoint);
      }
    };

    this.state = new QueueStateImpl(this);
  }

  public InternalPool getPool() {
    return pool;
  }

  public boolean isPrimaryUpdaterAlive() {
    boolean result = false;
    QueueConnectionImpl primary = (QueueConnectionImpl) queueConnections.getPrimary();
    if (primary != null) {
      ClientUpdater cu = primary.getUpdater();
      if (cu != null) {
        result = cu.isAlive();
      }
    }
    return result;
  }

  public QueueConnections getAllConnectionsNoWait() {
    return queueConnections;
  }

  public QueueConnections getAllConnections() {

    ConnectionList snapshot = queueConnections;
    if (snapshot.getPrimary() == null) {
      // wait for a new primary to become available.
      synchronized (lock) {
        snapshot = queueConnections;
        while (snapshot.getPrimary() == null && !snapshot.primaryDiscoveryFailed() && !shuttingDown
            && pool.getPoolOrCacheCancelInProgress() == null) {
          try {
            lock.wait();
          } catch (InterruptedException ignore) {
            Thread.currentThread().interrupt();
            break;
          }
          snapshot = queueConnections;
        }
      }
    }

    if (snapshot.getPrimary() == null) {
      pool.getCancelCriterion().checkCancelInProgress(null);
      GemFireException exception = snapshot.getPrimaryDiscoveryException();
      if (exception == null || exception instanceof NoSubscriptionServersAvailableException) {
        exception = new NoSubscriptionServersAvailableException(exception);
      } else {
        exception = new ServerConnectivityException(exception.getMessage(), exception);
      }
      throw exception;
    }

    return snapshot;
  }

  public InternalLogWriter getSecurityLogger() {
    return securityLogger;
  }

  public void close(boolean keepAlive) {
    endpointManager.removeListener(endpointListener);
    synchronized (lock) {
      shuttingDown = true;
      if (redundancySatisfierTask != null) {
        redundancySatisfierTask.cancel();
      }
      lock.notifyAll();
    }
    if (recoveryThread != null) {
      // it will be null if we never called start
      recoveryThread.shutdown();
    }
    if (recoveryThread != null) {
      try {
        if (!recoveryThread.awaitTermination(PoolImpl.SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS)) {
          logger.warn("Timeout waiting for recovery thread to complete");
        }
      } catch (InterruptedException ignore) {
        Thread.currentThread().interrupt();
        logger.debug("Interrupted waiting for recovery thread termination");
      }
    }

    QueueConnectionImpl primary = (QueueConnectionImpl) queueConnections.getPrimary();
    if (logger.isDebugEnabled()) {
      logger.debug("QueueManagerImpl - closing connections with keepAlive={}", keepAlive);
    }
    if (primary != null) {
      try {
        if (logger.isDebugEnabled()) {
          logger.debug("QueueManagerImpl - closing primary {}", primary);
        }
        primary.internalClose(keepAlive);
      } catch (Exception e) {
        logger.warn("Error closing primary connection to " +
            primary.getEndpoint(),
            e);
      }
    }

    List backups = queueConnections.getBackups();
    for (Iterator itr = backups.iterator(); itr.hasNext();) {
      QueueConnectionImpl backup = (QueueConnectionImpl) itr.next();
      if (backup != null) {
        try {
          if (logger.isDebugEnabled()) {
            logger.debug("QueueManagerImpl - closing backup {}", backup);
          }
          backup.internalClose(keepAlive);
        } catch (Exception e) {
          logger.warn("Error closing backup connection to " +
              backup.getEndpoint(),
              e);
        }
      }
    }
  }

  public void emergencyClose() {
    shuttingDown = true;
    queueConnections.getPrimary().emergencyClose();
    List backups = queueConnections.getBackups();
    for (int i = 0; i < backups.size(); i++) {
      Connection backup = (Connection) backups.get(i);
      backup.emergencyClose();
    }
  }

  public void start(ScheduledExecutorService background) {
    try {
      denyList.start(background);
      endpointManager.addListener(endpointListener);

      // Use a separate timer for queue management tasks
      // We don't want primary recovery (and therefore user threads) to wait for
      // things like pinging connections for health checks.
      final String name = "queueTimer-" + this.pool.getName();
      this.recoveryThread = LoggingExecutors.newScheduledThreadPool(name, 1, false);

      getState().start(background, getPool().getSubscriptionAckInterval());

      // initialize connections
      initializeConnections();

      scheduleRedundancySatisfierIfNeeded(redundancyRetryInterval);

      // When a server is removed from the denylist, try again
      // to establish redundancy (if we need to)
      DenyListListener denyListListener = new DenyListListenerAdapter() {
        @Override
        public void serverRemoved(ServerLocation location) {
          QueueManagerImpl.this.scheduleRedundancySatisfierIfNeeded(0);
        }
      };

      denyList.addListener(denyListListener);
      factory.getDenyList().addListener(denyListListener);
    } finally {
      initializedLatch.countDown();
    }
  }

  public void readyForEvents(InternalDistributedSystem system) {
    synchronized (lock) {
      this.sentClientReady = true;
    }

    QueueConnectionImpl primary = null;
    while (primary == null) {
      try {
        primary = (QueueConnectionImpl) getAllConnections().getPrimary();
      } catch (NoSubscriptionServersAvailableException ignore) {
        primary = null;
        break;
      }
      if (primary.sendClientReady()) {
        try {
          logger.info("Sending ready for events to primary: {}", primary);
          ReadyForEventsOp.execute(pool, primary);
        } catch (Exception e) {
          if (logger.isDebugEnabled()) {
            logger.debug("Error sending ready for events to {}", primary, e);
          }
          primary.destroy();
          primary = null;
        }
      }
    }
  }

  public void readyForEventsAfterFailover(QueueConnectionImpl primary) {
    try {
      logger.info("Sending ready for events to primary: {}", primary);
      ReadyForEventsOp.execute(pool, primary);
    } catch (Exception e) {
      if (logger.isDebugEnabled()) {
        logger.debug("Error sending ready for events to {}", primary, e);
      }
      primary.destroy();
    }
  }

  void connectionCrashed(Connection con) {
    // the endpoint has not crashed but this method does all the work
    // we need to do
    endpointCrashed(con.getEndpoint());
  }

  void endpointCrashed(Endpoint endpoint) {
    QueueConnectionImpl deadConnection = null;
    // We must be synchronized while checking to see if we have a queue connection for the endpoint,
    // because when we need to prevent a race between adding a queue connection to the map
    // and the endpoint for that connection crashing.
    synchronized (lock) {
      deadConnection = queueConnections.getConnection(endpoint);
      if (deadConnection != null) {
        queueConnections = queueConnections.removeConnection(deadConnection);
      }
    }
    if (deadConnection != null) {
      logger
          .info("{} subscription endpoint {} crashed. Scheduling recovery.",
              new Object[] {deadConnection.getUpdater() != null
                  ? (deadConnection.getUpdater().isPrimary() ? "Primary" : "Redundant")
                  : "Queue",
                  endpoint});
      scheduleRedundancySatisfierIfNeeded(0);
      deadConnection.internalDestroy();
    } else {
      if (logger.isDebugEnabled()) {
        logger.debug("Ignoring crashed endpoint {} it does not have a queue.", endpoint);
      }
    }
  }

  /**
   * This method checks whether queue connection exist on this endpoint or not. if its there then it
   * just destroys connection as clientUpdate thread is not there to read that connection.
   */
  public void checkEndpoint(ClientUpdater ccu, Endpoint endpoint) {
    QueueConnectionImpl deadConnection = null;

    synchronized (lock) {
      if (shuttingDown)
        return;
      // if same client updater then only remove as we don't know whether it has created new
      // updater/connection on same endpoint or not..
      deadConnection = queueConnections.getConnection(endpoint);
      if (deadConnection != null && ccu.equals(deadConnection.getUpdater())) {
        queueConnections = queueConnections.removeConnection(deadConnection);
        try {
          deadConnection.internalClose(pool.getKeepAlive());
        } catch (Exception e) {
          logger.warn("Error destroying client to server connection to {}",
              deadConnection.getEndpoint(), e);
        }
      }
    }

    logger
        .info("Cache client updater for {} on endpoint {} exiting. Scheduling recovery.",
            (deadConnection != null && deadConnection.getUpdater() != null)
                ? (deadConnection.getUpdater().isPrimary() ? "Primary" : "Redundant")
                : "Queue",
            endpoint);
    scheduleRedundancySatisfierIfNeeded(0);// one more chance
  }

  private void initializeConnections() {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      logger.debug("SubscriptionManager - intitializing connections");
    }

    int queuesNeeded = redundancyLevel == -1 ? -1 : redundancyLevel + 1;
    Set excludedServers = new HashSet(denyList.getBadServers());
    List servers = findQueueServers(excludedServers, queuesNeeded, true, false, null);

    if (servers == null || servers.isEmpty()) {
      logger.warn(
          "Could not create a queue. No queue servers available.");
      scheduleRedundancySatisfierIfNeeded(redundancyRetryInterval);
      synchronized (lock) {
        queueConnections = queueConnections.setPrimaryDiscoveryFailed(null);
        lock.notifyAll();
      }
      return;
    }

    if (isDebugEnabled) {
      logger.debug("SubscriptionManager - discovered subscription servers {}", servers);
    }

    SortedMap/* <ServerQueueStatus,Connection> */ oldQueueServers = new TreeMap(QSIZE_COMPARATOR);
    List nonRedundantServers = new ArrayList();

    for (Iterator itr = servers.iterator(); itr.hasNext();) {
      ServerLocation server = (ServerLocation) itr.next();
      Connection connection = null;
      try {
        connection = factory.createClientToServerConnection(server, true);
      } catch (GemFireSecurityException e) {
        throw e;
      } catch (GemFireConfigException e) {
        throw e;
      } catch (Exception e) {
        if (isDebugEnabled) {
          logger.debug("SubscriptionManager - Error connected to server: {}", server, e);
        }
      }
      if (connection != null) {
        ServerQueueStatus status = connection.getQueueStatus();
        if (status.isRedundant() || status.isPrimary()) {
          oldQueueServers.put(status, connection);
        } else {
          nonRedundantServers.add(connection);
        }
      }
    }

    // This ordering was determined from the old ConnectionProxyImpl code
    //
    // initialization order of the new redundant and primary server is
    // old redundant w/ second largest queue
    // old redundant w/ third largest queue
    // ...
    // old primary
    // non redundants in no particular order
    //
    // The primary is then chosen as
    // redundant with the largest queue
    // primary if there are no redundants
    // a non redundant

    // if the redundant with the largest queue fails, then we go and
    // make a new server a primary.

    Connection newPrimary = null;
    if (!oldQueueServers.isEmpty()) {
      newPrimary = (Connection) oldQueueServers.remove(oldQueueServers.lastKey());
    } else if (!nonRedundantServers.isEmpty()) {
      newPrimary = (Connection) nonRedundantServers.remove(0);
    }

    nonRedundantServers.addAll(0, oldQueueServers.values());

    for (Iterator itr = nonRedundantServers.iterator(); itr.hasNext();) {
      Connection connection = (Connection) itr.next();
      QueueConnectionImpl queueConnection = initializeQueueConnection(connection, false, null);
      if (queueConnection != null) {
        addToConnectionList(queueConnection, false);
      }
    }

    QueueConnectionImpl primaryQueue = null;
    if (newPrimary != null) {
      primaryQueue = initializeQueueConnection(newPrimary, true, null);
      if (primaryQueue == null) {
        newPrimary.destroy();
      } else {
        if (!addToConnectionList(primaryQueue, true)) {
          primaryQueue = null;
        }
      }
    }

    excludedServers.addAll(servers);

    // Make sure we have enough redundant copies. Some of the connections may
    // have failed
    // above.
    if (redundancyLevel != -1 && getCurrentRedundancy() < redundancyLevel) {
      if (isDebugEnabled) {
        logger.debug(
            "SubscriptionManager - Some initial connections failed. Trying to create redundant queues");
      }
      recoverRedundancy(excludedServers, false);
    }

    if (redundancyLevel != -1 && primaryQueue == null) {
      if (isDebugEnabled) {
        logger.debug(
            "SubscriptionManager - Intial primary creation failed. Trying to create a new primary");
      }
      while (primaryQueue == null) {
        primaryQueue = createNewPrimary(excludedServers);
        if (primaryQueue == null) {
          // couldn't find a server to make primary
          break;
        }
        if (!addToConnectionList(primaryQueue, true)) {
          excludedServers.add(primaryQueue.getServer());
          primaryQueue = null;
        }
      }
    }

    if (primaryQueue == null) {
      if (isDebugEnabled) {
        logger.debug(
            "SubscriptionManager - Unable to create a new primary queue, using one of the redundant queues");
      }
      while (primaryQueue == null) {
        primaryQueue = promoteBackupToPrimary(queueConnections.getBackups());
        if (primaryQueue == null) {
          // no backup servers available
          break;
        }
        if (!addToConnectionList(primaryQueue, true)) {
          synchronized (lock) {
            // make sure we don't retry this same connection.
            queueConnections = queueConnections.removeConnection(primaryQueue);
          }
          primaryQueue = null;
        }
      }
    }

    if (primaryQueue == null) {
      logger.error("Could not initialize a primary queue on startup. No queue servers available.");
      synchronized (lock) {
        queueConnections =
            queueConnections.setPrimaryDiscoveryFailed(new NoSubscriptionServersAvailableException(
                "Could not initialize a primary queue on startup. No queue servers available."));
        lock.notifyAll();
      }
      cqsDisconnected();
    } else {
      cqsConnected();
    }

    if (getCurrentRedundancy() < redundancyLevel) {
      logger.warn(
          "Unable to initialize enough redundant queues on startup. The redundancy count is currently {}.",
          getCurrentRedundancy());
    }
  }

  private void cqsConnected() {
    InternalCache cache = GemFireCacheImpl.getInstance();
    if (cache != null) {
      CqService cqService = cache.getCqService();
      // Primary queue was found, alert the affected cqs if necessary
      cqService.cqsConnected(pool);
    }
  }

  private void cqsDisconnected() {
    // No primary queue was found, alert the affected cqs if necessary
    InternalCache cache = GemFireCacheImpl.getInstance();
    if (cache != null) {
      CqService cqService = cache.getCqService();
      cqService.cqsDisconnected(pool);
    }
  }

  private int getCurrentRedundancy() {
    return queueConnections.getBackups().size();
  }

  /**
   * Make sure that we have enough backup servers.
   *
   * Add any servers we fail to connect to to the excluded servers list.
   */
  protected boolean recoverRedundancy(Set excludedServers, boolean recoverInterest) {
    if (pool.getPoolOrCacheCancelInProgress() != null) {
      return true;
    }
    int additionalBackups;
    while (pool.getPoolOrCacheCancelInProgress() == null
        && ((additionalBackups = redundancyLevel - getCurrentRedundancy()) > 0
            || redundancyLevel == -1)) {

      if (redundancyLevel != -1 && printRecoveringRedundant) {
        logger.info(
            "SubscriptionManager redundancy satisfier - redundant endpoint has been lost. Attempting to recover.");
        printRecoveringRedundant = false;
      }

      List servers = findQueueServers(excludedServers,
          redundancyLevel == -1 ? -1 : additionalBackups, false,
          (redundancyLevel == -1 ? false : printRedundancyNotSatisfiedError),
          "Could not find any server to host redundant client queue. Number of excluded servers is %s and exception is %s");

      if (servers == null || servers.isEmpty()) {
        if (redundancyLevel != -1) {

          if (printRedundancyNotSatisfiedError) {
            logger.info(
                "Redundancy level {} is not satisfied, but there are no more servers available. Redundancy is currently {}.",
                new Object[] {redundancyLevel, getCurrentRedundancy()});
          }
        }
        printRedundancyNotSatisfiedError = false;// printed above
        return false;
      }
      excludedServers.addAll(servers);

      final boolean isDebugEnabled = logger.isDebugEnabled();
      for (Iterator itr = servers.iterator(); itr.hasNext();) {
        ServerLocation server = (ServerLocation) itr.next();
        Connection connection = null;
        try {
          connection = factory.createClientToServerConnection(server, true);
        } catch (GemFireSecurityException e) {
          throw e;
        } catch (Exception e) {
          if (isDebugEnabled) {
            logger.debug("SubscriptionManager - Error connecting to server: ()", server, e);
          }
        }
        if (connection == null) {
          continue;
        }

        QueueConnectionImpl queueConnection = initializeQueueConnection(connection, false, null);
        if (queueConnection != null) {
          boolean isFirstNewConnection = false;
          synchronized (lock) {
            if (recoverInterest && queueConnections.getPrimary() == null
                && queueConnections.getBackups().isEmpty()) {
              // we lost our queue at some point. We Need to recover
              // interest. This server will be made primary after this method
              // finishes
              // because whoever killed the primary when this method started
              // should
              // have scheduled a task to recover the primary.
              isFirstNewConnection = true;
              // TODO - Actually, we need a better check than the above. There's
              // still a chance
              // that we haven't realized that the primary has died but it is
              // already gone. We should
              // get some information from the queue server about whether it was
              // able to copy the
              // queue from another server and decide if we need to recover our
              // interest based on
              // that information.
            }
          }
          boolean promotionFailed = false;
          if (isFirstNewConnection) {
            if (!promoteBackupCnxToPrimary(queueConnection)) {
              promotionFailed = true;
            }
          }
          if (!promotionFailed) {
            if (addToConnectionList(queueConnection, isFirstNewConnection)) {
              // redundancy satisfied
              printRedundancyNotSatisfiedError = true;
              printRecoveringRedundant = true;
              if (logger.isDebugEnabled()) {
                logger.debug(
                    "SubscriptionManager redundancy satisfier - created a queue on server {}",
                    queueConnection.getEndpoint());
              }
              // Even though the new redundant queue will usually recover
              // subscription information (see bug #39014) from its initial
              // image provider, in bug #42280 we found that this is not always
              // the case, so clients must always register interest with the new
              // redundant server.
              if (recoverInterest) {
                recoverInterest(queueConnection, isFirstNewConnection);
              }
            }
          }
        }
      }
    }
    return true;
  }

  private QueueConnectionImpl promoteBackupToPrimary(List backups) {
    QueueConnectionImpl primary = null;
    for (int i = 0; primary == null && i < backups.size(); i++) {
      QueueConnectionImpl lastConnection = (QueueConnectionImpl) backups.get(i);
      if (promoteBackupCnxToPrimary(lastConnection)) {
        primary = lastConnection;
      }
    }
    return primary;
  }

  private boolean promoteBackupCnxToPrimary(QueueConnectionImpl cnx) {
    boolean result = false;
    if (PoolImpl.BEFORE_PRIMARY_IDENTIFICATION_FROM_BACKUP_CALLBACK_FLAG) {
      ClientServerObserver bo = ClientServerObserverHolder.getInstance();
      bo.beforePrimaryIdentificationFromBackup();
    }
    try {
      boolean haveSentClientReady = this.sentClientReady;
      if (haveSentClientReady) {
        cnx.sendClientReady();
      }
      ClientUpdater updater = cnx.getUpdater();
      if (updater == null) {
        if (logger.isDebugEnabled()) {
          logger.debug("backup connection was destroyed before it could become the primary.");
        }
        Assert.assertTrue(cnx.isDestroyed());
      } else {
        updater.setFailedUpdater(queueConnections.getFailedUpdater());
        MakePrimaryOp.execute(pool, cnx, haveSentClientReady);
        result = true;
        if (PoolImpl.AFTER_PRIMARY_IDENTIFICATION_FROM_BACKUP_CALLBACK_FLAG) {
          ClientServerObserver bo = ClientServerObserverHolder.getInstance();
          bo.afterPrimaryIdentificationFromBackup(cnx.getServer());
        }
      }
    } catch (Exception e) {
      if (pool.getPoolOrCacheCancelInProgress() == null && logger.isDebugEnabled()) {
        logger.debug("Error making a backup server the primary server for client subscriptions", e);
      }
    }
    return result;
  }

  /**
   * Create a new primary server from a non-redundant server.
   *
   * Add any failed servers to the excludedServers set.
   */
  private QueueConnectionImpl createNewPrimary(Set excludedServers) {
    QueueConnectionImpl primary = null;
    while (primary == null && pool.getPoolOrCacheCancelInProgress() == null) {
      List servers = findQueueServers(excludedServers, 1, false, printPrimaryNotFoundError,
          "Could not find any server to host primary client queue. Number of excluded servers is %s and exception is %s");
      printPrimaryNotFoundError = false; // printed above
      if (servers == null || servers.isEmpty()) {
        break;
      }

      Connection connection = null;
      try {
        connection = factory.createClientToServerConnection((ServerLocation) servers.get(0), true);
      } catch (GemFireSecurityException e) {
        throw e;
      } catch (Exception e) {
        if (logger.isDebugEnabled()) {
          logger.debug("SubscriptionManagerImpl - error creating a connection to server {}",
              servers.get(0));
        }
      }
      if (connection != null) {
        primary = initializeQueueConnection(connection, true, queueConnections.getFailedUpdater());
      }
      excludedServers.addAll(servers);
    }

    if (primary != null && sentClientReady && primary.sendClientReady()) {
      readyForEventsAfterFailover(primary);
    }
    return primary;
  }

  private List findQueueServers(Set excludedServers, int count, boolean findDurable,
      boolean printErrorMessage, String msg) {
    List servers = null;
    Exception ex = null;
    try {
      if (pool.getPoolOrCacheCancelInProgress() != null) {
        return null;
      }
      servers = source.findServersForQueue(excludedServers, count, proxyId, findDurable);
    } catch (GemFireSecurityException e) {
      // propagate the security exception immediately.
      throw e;
    } catch (Exception e) {
      /*
       * logger .warning(
       * LocalizedStrings.QueueManagerImpl_COULD_NOT_RETRIEVE_LIST_OF_SERVERS_FOR_SUBSCRIPTION_0,
       * new Object[] { e.getMessage() });
       */
      ex = e;
      if (logger.isDebugEnabled()) {
        logger.debug("SubscriptionManager - Error getting the list of servers: {}", e);
      }
    }

    if (printErrorMessage) {
      if (servers == null || servers.isEmpty()) {
        logger.error(String.format(msg,
            new Object[] {(excludedServers != null ? excludedServers.size() : 0),
                (ex != null ? ex.getMessage() : "no exception")}));
      }
    }
    return servers;
  }

  /**
   * Find a new primary, adding any failed servers we encounter to the excluded servers list
   *
   * First we try to make a backup server the primary, but if run out of backup servers we will try
   * to find a new server.
   */
  protected void recoverPrimary(Set excludedServers) {
    if (pool.getPoolOrCacheCancelInProgress() != null) {
      return;
    }
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (queueConnections.getPrimary() != null) {
      if (isDebugEnabled) {
        logger.debug("Primary recovery not needed");
      }
      return;
    }

    if (isDebugEnabled) {
      logger.debug(
          "SubscriptionManager redundancy satisfier - primary endpoint has been lost. Attempting to recover");
    }

    if (printRecoveringPrimary) {
      logger.info(
          "SubscriptionManager redundancy satisfier - primary endpoint has been lost. Attempting to recover.");
      printRecoveringPrimary = false;
    }

    QueueConnectionImpl newPrimary = null;
    while (newPrimary == null && pool.getPoolOrCacheCancelInProgress() == null) {
      List backups = queueConnections.getBackups();
      newPrimary = promoteBackupToPrimary(backups);
      // Hitesh now lets say that server crashed
      if (newPrimary == null) {
        // could not find a backup to promote
        break;
      }
      if (!addToConnectionList(newPrimary, true)) {
        synchronized (lock) {
          // make sure we don't retry the same backup server
          queueConnections = queueConnections.removeConnection(newPrimary);
        }
        newPrimary = null;
      }

    }

    if (newPrimary != null) {
      if (isDebugEnabled) {
        logger.debug(
            "SubscriptionManager redundancy satisfier - Switched backup server to primary: {}",
            newPrimary.getEndpoint());
      }
      if (PoolImpl.AFTER_PRIMARY_RECOVERED_CALLBACK_FLAG) {
        ClientServerObserver bo = ClientServerObserverHolder.getInstance();
        bo.afterPrimaryRecovered(newPrimary.getServer());
      }

      // new primary from back up server was found, alert affected cqs if necessary
      cqsConnected();
      printPrimaryNotFoundError = true;
      printRecoveringPrimary = true;
      return;
    }

    while (newPrimary == null) {
      newPrimary = createNewPrimary(excludedServers);
      if (newPrimary == null) {
        // could not find a new primary to create
        break;
      }
      if (!addToConnectionList(newPrimary, true)) {
        excludedServers.add(newPrimary.getServer());
        newPrimary = null;
      }

      if (newPrimary != null) {
        if (isDebugEnabled) {
          logger.debug(
              "SubscriptionManager redundancy satisfier - Non backup server was made primary. Recovering interest {}",
              newPrimary.getEndpoint());
        }

        if (!recoverInterest(newPrimary, true)) {
          excludedServers.add(newPrimary.getServer());
          newPrimary = null;
        }
        // New primary queue was found from a non backup, alert the affected cqs
        cqsConnected();
      }

      if (newPrimary != null && PoolImpl.AFTER_PRIMARY_RECOVERED_CALLBACK_FLAG) {
        ClientServerObserver bo = ClientServerObserverHolder.getInstance();
        bo.afterPrimaryRecovered(newPrimary.getServer());
      }
      printPrimaryNotFoundError = true;
      printRecoveringPrimary = true;
      return;
    }
    // No primary queue was found, alert the affected cqs
    cqsDisconnected();
    if (isDebugEnabled) {
      logger.debug("SubscriptionManager redundancy satisfier - Could not recover a new primary");
    }
    synchronized (lock) {
      queueConnections = queueConnections.setPrimaryDiscoveryFailed(null);
      lock.notifyAll();
    }
  }

  private QueueConnectionImpl initializeQueueConnection(Connection connection, boolean isPrimary,
      ClientUpdater failedUpdater) {
    QueueConnectionImpl queueConnection = null;
    FailureTracker failureTracker = denyList.getFailureTracker(connection.getServer());
    try {
      ClientUpdater updater = factory.createServerToClientConnection(connection.getEndpoint(), this,
          isPrimary, failedUpdater);
      if (updater != null) {
        queueConnection = new QueueConnectionImpl(this, connection, updater, failureTracker);
      } else {
        logger.warn("unable to create a subscription connection to server {}",
            connection.getEndpoint());
      }
    } catch (Exception e) {
      if (logger.isDebugEnabled()) {
        logger.debug("error creating subscription connection to server {}",
            connection.getEndpoint(), e);
      }
    }
    if (queueConnection == null) {
      failureTracker.addFailure();
      connection.destroy();
    }
    return queueConnection;
  }

  // need flag whether primary is created from backup
  // for backuup queue lets assume before we add connection, endpoint crashed, now we put in
  // connection but CCU may died as endpoint closed....
  // so before putting connection need to see if something(crash) happen we should be able to
  // recover from it
  private boolean addToConnectionList(QueueConnectionImpl connection, boolean isPrimary) {
    boolean isBadConnection;
    synchronized (lock) {
      ClientUpdater cu = connection.getUpdater();
      if (cu == null || (!cu.isAlive()) || (!cu.isProcessing()))
        return false;// don't add
      // now still CCU can died but then it will execute Checkendpoint with lock it will remove
      // connection connection and it will reschedule it.
      if (connection.getEndpoint().isClosed() || shuttingDown
          || pool.getPoolOrCacheCancelInProgress() != null) {
        isBadConnection = true;
      } else {
        isBadConnection = false;
        if (isPrimary) {
          queueConnections = queueConnections.setPrimary(connection);
          lock.notifyAll();
        } else {
          queueConnections = queueConnections.addBackup(connection);
        }
      }
    }

    if (isBadConnection) {
      if (logger.isDebugEnabled()) {
        logger.debug(
            "Endpoint {} crashed while creating a connection. The connection will be destroyed",
            connection.getEndpoint());
      }
      try {
        connection.internalClose(pool.getKeepAlive());
      } catch (Exception e) {
        if (logger.isDebugEnabled()) {
          logger.debug("Error destroying client to server connection to {}",
              connection.getEndpoint(), e);
        }
      }
    }

    return !isBadConnection;
  }

  protected void scheduleRedundancySatisfierIfNeeded(long delay) {
    if (shuttingDown) {
      return;
    }

    synchronized (lock) {
      if (shuttingDown) {
        return;
      }
      if (queueConnections.getPrimary() == null || getCurrentRedundancy() < redundancyLevel
          || redundancyLevel == -1 || queueConnections.primaryDiscoveryFailed()) {
        if (redundancySatisfierTask != null) {
          if (redundancySatisfierTask.getRemainingDelay() > delay) {
            redundancySatisfierTask.cancel();
          } else {
            return;
          }
        }

        redundancySatisfierTask = new RedundancySatisfierTask();
        try {
          ScheduledFuture future =
              recoveryThread.schedule(redundancySatisfierTask, delay, TimeUnit.MILLISECONDS);
          redundancySatisfierTask.setFuture(future);
        } catch (RejectedExecutionException e) {
          // ignore, the timer has been cancelled, which means we're shutting down.
        }
      }
    }
  }

  private boolean recoverInterest(final QueueConnectionImpl newConnection,
      final boolean isFirstNewConnection) {

    if (pool.getPoolOrCacheCancelInProgress() != null) {
      return true;
    }

    // recover interest
    try {
      recoverAllInterestTypes(newConnection, isFirstNewConnection);
      newConnection.getFailureTracker().reset();
      return true;
    } catch (CancelException ignore) {
      return true;
      // ok to ignore we are being shutdown
    } 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) {
      SystemFailure.checkFailure();
      pool.getCancelCriterion().checkCancelInProgress(t);
      logger.warn("QueueManagerImpl failed to recover interest to server " +
          newConnection.getServer(),
          t);
      newConnection.getFailureTracker().addFailure();
      newConnection.destroy();
      return false;
    }
  }

  public QueueState getState() {
    return this.state;
  }

  private void recoverSingleList(int interestType, Connection recoveredConnection,
      boolean isDurable, boolean receiveValues, boolean isFirstNewConnection) {
    Iterator i = this.getPool().getRITracker()
        .getRegionToInterestsMap(interestType, isDurable, !receiveValues).values().iterator();
    while (i.hasNext()) { // restore a region
      RegionInterestEntry e = (RegionInterestEntry) i.next();
      recoverSingleRegion(e.getRegion(), e.getInterests(), interestType, recoveredConnection,
          isDurable, receiveValues, isFirstNewConnection);
    } // restore a region
  }

  private void recoverCqs(Connection recoveredConnection, boolean isDurable) {
    Map cqs = this.getPool().getRITracker().getCqsMap();
    Iterator i = cqs.entrySet().iterator();
    while (i.hasNext()) {
      Map.Entry e = (Map.Entry) i.next();
      ClientCQ cqi = (ClientCQ) e.getKey();
      String name = cqi.getName();
      if (this.pool.getMultiuserAuthentication()) {
        UserAttributes.userAttributes
            .set(((DefaultQueryService) this.pool.getQueryService()).getUserAttributes(name));
      }
      try {
        if (((CqStateImpl) cqi.getState()).getState() != CqStateImpl.INIT) {
          cqi.createOn(recoveredConnection, isDurable);
        }
      } finally {
        UserAttributes.userAttributes.set(null);
      }
    }
  }

  // TODO this is distressingly similar to LocalRegion#processSingleInterest
  private void recoverSingleRegion(LocalRegion r, Map keys, int interestType,
      Connection recoveredConnection, boolean isDurable, boolean receiveValues,
      boolean isFirstNewConnection) {

    if (logger.isDebugEnabled()) {
      logger.debug("{}.recoverSingleRegion starting kind={} region={}: {}", this,
          InterestType.getString(interestType), r.getFullPath(), keys);
    }

    // build a HashMap, key is policy, value is list
    HashMap policyMap = new HashMap();
    Iterator keysIter = keys.entrySet().iterator();
    while (keysIter.hasNext()) { // restore and commit an interest
      Map.Entry me = (Map.Entry) keysIter.next();
      Object key = me.getKey();
      InterestResultPolicy pol = (InterestResultPolicy) me.getValue();

      if (interestType == InterestType.KEY) {
        // Gester: we only consolidate the key into list for InterestType.KEY
        LinkedList keyList = (LinkedList) policyMap.get(pol);
        if (keyList == null) {

          keyList = new LinkedList();
        }
        keyList.add(key);
        policyMap.put(pol, keyList);
      } else {
        // for other Interest type, do it one by one
        recoverSingleKey(r, key, pol, interestType, recoveredConnection, isDurable, receiveValues,
            isFirstNewConnection);
      }
    }

    // Process InterestType.KEY: Iterator list for each each policy
    Iterator polIter = policyMap.entrySet().iterator();
    while (polIter.hasNext()) {
      Map.Entry me = (Map.Entry) polIter.next();
      LinkedList keyList = (LinkedList) me.getValue();
      InterestResultPolicy pol = (InterestResultPolicy) me.getKey();
      recoverSingleKey(r, keyList, pol, interestType, recoveredConnection, isDurable, receiveValues,
          isFirstNewConnection);
    }
  }

  private void recoverSingleKey(LocalRegion r, Object keys, InterestResultPolicy policy,
      int interestType, Connection recoveredConnection, boolean isDurable, boolean receiveValues,
      boolean isFirstNewConnection) {
    r.startRegisterInterest();
    try {
      // Remove all matching values from local cache
      if (isFirstNewConnection) { // only if this recoveredEP
        // becomes primaryEndpoint
        r.clearKeysOfInterest(keys, interestType, policy);
        if (logger.isDebugEnabled()) {
          logger.debug(
              "{}.recoverSingleRegion :Endpoint recovered is primary so clearing the keys of interest starting kind={} region={}: {}",
              this, InterestType.getString(interestType), r.getFullPath(), keys);
        }
      }
      // Register interest, get new values back
      List serverKeys;
      if (policy != InterestResultPolicy.KEYS_VALUES) {
        serverKeys = r.getServerProxy().registerInterestOn(recoveredConnection, keys, interestType,
            policy, isDurable, !receiveValues, r.getAttributes().getDataPolicy().ordinal);
        // Restore keys based on server's response
        if (isFirstNewConnection) {
          // only if this recoveredEP becomes primaryEndpoint
          r.refreshEntriesFromServerKeys(recoveredConnection, serverKeys, policy);
        }
      } else {
        if (!isFirstNewConnection) {
          // InterestResultPolicy.KEYS_VALUES now fetches values in
          // RegisterInterestOp's response itself and in this case
          // refreshEntriesFromServerKeys(...) does not explicitly fetch values
          // but only updates keys-values to region. To not fetch values, we
          // need to use policy NONE or KEYS.
          serverKeys = r.getServerProxy().registerInterestOn(recoveredConnection, keys,
              interestType, InterestResultPolicy.NONE, isDurable, !receiveValues,
              r.getAttributes().getDataPolicy().ordinal);
        } else {
          serverKeys =
              r.getServerProxy().registerInterestOn(recoveredConnection, keys, interestType, policy,
                  isDurable, !receiveValues, r.getAttributes().getDataPolicy().ordinal);
          r.refreshEntriesFromServerKeys(recoveredConnection, serverKeys, policy);
        }
      }
    } finally {
      r.finishRegisterInterest();
    }
  }

  private void recoverInterestList(final Connection recoveredConnection, boolean durable,
      boolean receiveValues, boolean isFirstNewConnection) {
    recoverSingleList(InterestType.KEY, recoveredConnection, durable, receiveValues,
        isFirstNewConnection);
    recoverSingleList(InterestType.REGULAR_EXPRESSION, recoveredConnection, durable, receiveValues,
        isFirstNewConnection);
    recoverSingleList(InterestType.FILTER_CLASS, recoveredConnection, durable, receiveValues,
        isFirstNewConnection);
    recoverSingleList(InterestType.OQL_QUERY, recoveredConnection, durable, receiveValues,
        isFirstNewConnection);
  }

  protected void recoverAllInterestTypes(final Connection recoveredConnection,
      boolean isFirstNewConnection) {
    if (PoolImpl.BEFORE_RECOVER_INTEREST_CALLBACK_FLAG) {
      ClientServerObserver bo = ClientServerObserverHolder.getInstance();
      bo.beforeInterestRecovery();
    }
    recoverInterestList(recoveredConnection, false, true, isFirstNewConnection);
    recoverInterestList(recoveredConnection, false, false, isFirstNewConnection);
    recoverCqs(recoveredConnection, false);
    if (getPool().isDurableClient()) {
      recoverInterestList(recoveredConnection, true, true, isFirstNewConnection);
      recoverInterestList(recoveredConnection, true, false, isFirstNewConnection);
      recoverCqs(recoveredConnection, true);
    }
  }

  /**
   * A comparator which sorts queue elements in the order of primary first redundant with smallest
   * queue size ... redundant with largest queue size
   *
   *
   */
  protected static class QSizeComparator implements java.util.Comparator {
    public int compare(Object o1, Object o2) {
      ServerQueueStatus s1 = (ServerQueueStatus) o1;
      ServerQueueStatus s2 = (ServerQueueStatus) o2;
      // sort primaries to the front of the list
      if (s1.isPrimary() && !s2.isPrimary()) {
        return -1;
      } else if (!s1.isPrimary() && s2.isPrimary()) {
        return 1;
      } else {
        int diff = s1.getServerQueueSize() - s2.getServerQueueSize();
        if (diff != 0) {
          return diff;
        } else {
          return s1.getMemberId().compareTo(s2.getMemberId());
        }
      }
    }
  }

  /**
   * A data structure for holding the current set of connections the queueConnections reference
   * should be maintained by making a copy of this data structure for each change.
   *
   * Note the the order of the backups is significant. The first backup in the list is the first
   * server that will be become primary after the primary fails, etc.
   *
   * The order of backups in this list is the reverse of the order or endpoints from the old
   * ConnectionProxyImpl .
   */
  public class ConnectionList implements QueueConnections {
    private final QueueConnectionImpl primary;
    private final Map/* <Endpoint, QueueConnection> */ connectionMap;
    private final List/* <QueueConnection> */ backups;
    /**
     * The primaryDiscoveryException flag is stronger than just not having any queue connections It
     * also means we tried all of the possible queue servers and we'ren't able to connect.
     */
    private final GemFireException primaryDiscoveryException;
    private final QueueConnectionImpl failedPrimary;

    public ConnectionList() {
      primary = null;
      connectionMap = Collections.EMPTY_MAP;
      backups = Collections.EMPTY_LIST;
      primaryDiscoveryException = null;
      failedPrimary = null;
    }

    private ConnectionList(QueueConnectionImpl primary, List backups,
        GemFireException discoveryException, QueueConnectionImpl failedPrimary) {
      this.primary = primary;
      Map allConnectionsTmp = new HashMap();
      for (Iterator itr = backups.iterator(); itr.hasNext();) {
        QueueConnectionImpl nextConnection = (QueueConnectionImpl) itr.next();
        allConnectionsTmp.put(nextConnection.getEndpoint(), nextConnection);
      }
      if (primary != null) {
        allConnectionsTmp.put(primary.getEndpoint(), primary);
      }
      this.connectionMap = Collections.unmodifiableMap(allConnectionsTmp);
      this.backups = Collections.unmodifiableList(new ArrayList(backups));
      pool.getStats().setSubscriptionCount(connectionMap.size());
      this.primaryDiscoveryException = discoveryException;
      this.failedPrimary = failedPrimary;
    }

    public ConnectionList setPrimary(QueueConnectionImpl newPrimary) {
      List newBackups = backups;
      if (backups.contains(newPrimary)) {
        newBackups = new ArrayList(backups);
        newBackups.remove(newPrimary);
      }
      return new ConnectionList(newPrimary, newBackups, null, null);
    }

    public ConnectionList setPrimaryDiscoveryFailed(GemFireException p_discoveryException) {
      GemFireException discoveryException = p_discoveryException;
      if (discoveryException == null) {
        discoveryException =
            new NoSubscriptionServersAvailableException("Primary discovery failed.");
      }
      return new ConnectionList(primary, backups, discoveryException, failedPrimary);
    }

    public ConnectionList addBackup(QueueConnectionImpl queueConnection) {
      ArrayList newBackups = new ArrayList(backups);
      newBackups.add(queueConnection);
      return new ConnectionList(primary, newBackups, primaryDiscoveryException, failedPrimary);
    }

    public ConnectionList removeConnection(QueueConnectionImpl connection) {
      if (primary == connection) {
        return new ConnectionList(null, backups, primaryDiscoveryException, primary);
      } else {
        ArrayList newBackups = new ArrayList(backups);
        newBackups.remove(connection);
        return new ConnectionList(primary, newBackups, primaryDiscoveryException, failedPrimary);
      }
    }

    public Connection getPrimary() {
      return primary;
    }

    public List/* <QueueConnection> */ getBackups() {
      return backups;
    }

    /**
     * Return the cache client updater from the previously failed primary
     *
     * @return the previous updater or null if there is no previous updater
     */
    public ClientUpdater getFailedUpdater() {
      if (failedPrimary != null) {
        return failedPrimary.getUpdater();
      } else {
        return null;
      }
    }

    public boolean primaryDiscoveryFailed() {
      return primaryDiscoveryException != null;
    }

    public GemFireException getPrimaryDiscoveryException() {
      return primaryDiscoveryException;
    }

    public QueueConnectionImpl getConnection(Endpoint endpoint) {
      return (QueueConnectionImpl) connectionMap.get(endpoint);
    }

    /** return a copy of the list of all server locations */
    public Set/* <ServerLocation> */ getAllLocations() {
      HashSet locations = new HashSet();
      for (Iterator itr = connectionMap.keySet().iterator(); itr.hasNext();) {
        org.apache.geode.cache.client.internal.Endpoint endpoint =
            (org.apache.geode.cache.client.internal.Endpoint) itr.next();
        locations.add(endpoint.getLocation());
      }

      return locations;
    }
  }

  protected void logError(String message, Throwable t) {
    if (t instanceof GemFireSecurityException) {
      securityLogger.error(message, t);
    } else {
      logger.error(message, t);
    }
  }

  /**
   * Asynchronous task which tries to restablish a primary connection and satisfy redundant
   * requirements.
   *
   * This task should only be running in a single thread at a time. This task is the only way that
   * new queue servers will be added, and the only way that a backup server can transistion to a
   * primary server.
   *
   */
  protected class RedundancySatisfierTask extends PoolTask {
    private boolean isCancelled;
    private ScheduledFuture future;

    public void setFuture(ScheduledFuture future) {
      this.future = future;
    }

    public long getRemainingDelay() {
      return future.getDelay(TimeUnit.MILLISECONDS);
    }

    @Override
    public void run2() {
      try {
        initializedLatch.await();
        synchronized (lock) {
          if (isCancelled) {
            return;
          } else {
            redundancySatisfierTask = null;
          }
          if (pool.getPoolOrCacheCancelInProgress() != null) {
            /* wake up waiters so they can detect cancel */
            lock.notifyAll();
            return;
          }
        }
        Set excludedServers = queueConnections.getAllLocations();
        excludedServers.addAll(denyList.getBadServers());
        excludedServers.addAll(factory.getDenyList().getBadServers());
        recoverPrimary(excludedServers);
        recoverRedundancy(excludedServers, true);
      } 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 (CancelException e) {
        throw e;
      } 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();
        synchronized (lock) {
          if (t instanceof GemFireSecurityException) {
            queueConnections =
                queueConnections.setPrimaryDiscoveryFailed((GemFireSecurityException) t);
          } else {
            queueConnections = queueConnections.setPrimaryDiscoveryFailed(null);
          }
          lock.notifyAll();
          pool.getCancelCriterion().checkCancelInProgress(t);
          logError("Error in redundancy satisfier", t);
        }
      }

      scheduleRedundancySatisfierIfNeeded(redundancyRetryInterval);
    }

    public boolean cancel() {
      synchronized (lock) {
        if (isCancelled) {
          return false;
        }
        isCancelled = true;
        future.cancel(false);
        redundancySatisfierTask = null;
        return true;
      }
    }

  }

  public static void loadEmergencyClasses() {
    QueueConnectionImpl.loadEmergencyClasses();
  }
}

redpajama_set_name: RedPajamaGithub


The Alan Bible Visitor Center, built in 1966 has been brought into the 21st century while preserving its historic elements. Improvements include bringing its many systems up to code, improving accessibility throughout the building and making it more climate friendly by conserving and producing energy.
Our Visitor Center volunteers can answer all your park questions and they have a wealth of knowledge about the park, it's history, and things to do.
The Visitor Center is located outside the park entrance (no fee required) near Boulder City.
Those driving by the visitor center site will notice the new solar panel array on the property. These nine panels and other improvements will earn the Alan Bible Visitor Center a LEED (Leadership in Energy and Environmental Design) silver rating, a unique award for a historic building.
Other changes are more subtle; an exciting new award winning park film, the improved bookstore carries many new educational and nature gift items and the cactus garden will have improved sidewalks and masonry walls.
See a full size relief map of the Lake Mead National Recreation Area. This map shows the park in amazing detail.
Explore what the park has to offer from what animals make Lake Mead their home to the challenges the park faces into the future.
The Alan Bible Visitor Center also offers a complete gift store stocked with books about the park and the region, posters, history, T-shirts, hats, postcards, and maps. If you have any questions about the Lake Mead National Recreation Area this is the place to find information on anything related to the park.
The Visitor Center is open 9:00 a.m. to 4:30 p.m daily.
Please call 702-293-8990 for more information.
The Visitor Center will be closed the following holidays: Thanksgiving Day, Christmas Day and New Years Day.


text:
class Cache(object):

    def __init__(self):
        self._cache = {}

    def get(self, key):
        """
        Get the value stored for the key or None if it does not exist
        :param key:
        :return:
        """
        return self._cache.get(key, None)

    def put(self, key, value):
        """
        Store a value for the key. If it does exist, it will be silently overwritten.
        :param key:
        :param value:
        :return:
        """
        self._cache[key] = value


text:
This site is owned by the company WEG Equipamentos Elétricos S.A. ("we", "us", "our" or "WEG").
By using the Site, the User declares to have read and agreed to the rules described in this document, becoming entitled to access the services that the site provides. If you do not accept this Terms and Conditions of Use, you will not be allowed access to the restricted areas of the Site.
WEG reserves the right to update or modify this Terms of Use. The most recent version eliminates all previous ones, in this way, the user is responsible for checking the Terms whenever they access the Site.
If you have any questions or concerns, please mail through the "Contact Us" section of www.weg.net
2. DATA PROTECTION AND PRIVACY
By collecting information, we aim to provide our users with a high-quality service and an efficient and personalized experience while using the site.
WEG sets out the rules for the collection, use, processing and storage of the User's personal data provided by it for use of the Site services. For more information about how WEG addresses the privacy of personal data, please read our privacy policy.
We use cookies to optimize site (s) and analyze website traffic. Cookies are an essential part of how our site works. The main purpose of our cookies is to improve your browsing experience. For example, they are used to remember your preferences (language, country, etc.) while browsing and on future visits.
The information collected by Cookies also allows us to improve the site, estimating numbers and usage patterns, the suitability of the site to the individual interests of the users, faster searches, etc.
At any time, you can change your browser settings to manage these cookies. For more information on how WEG uses cookies, see our cookie statement.
Upon release of your registration, the site will grant the user a personal "login and password" that will correspond to your identification next to the site, which will allow you to access the services and features. You are responsible for protecting the confidentiality of your personal password.
Logins that have been previously chosen by another user will not be accepted.
The user declares and acknowledges that the information provided by him is true, correct, current and complete, being responsible for this information. If the data entered at the time of registration are incorrect or incomplete, making it impossible to prove and identify the user, the site will have the right to suspend immediately the provision of services, without the need for prior notice.
5. REGISTRATION DATA
The user declares to be aware that, from his registration in the site, his information will become part of a database, being able the user to receive information and informative contents (new functionalities). If the user does not have an interest in receiving informative content, he is assured the right to express his opposition, if such a decision is communicated by the channel "Contact Us".
6. TEMPORARY INTERRUPTION OF SERVICES
While WEG makes every effort to keep the site up and running for twenty-four (24) hours per day, seven (7) days per week, there may be times when it is unavailable for different reasons, including virtue of performing routine maintenance. In such situations, access may be interrupted, suspended or terminated by WEG in its sole discretion.
7. PURCHASE OF PRODUCTS
By opting for the purchase through the WEG website, the user will be obliged to comply with the signed terms, being obliged to respect the terms of payment and / or termination and / or repentance, according to the law, as well as tacitly agreeing with the respective Terms of Use of the company's website.
8. INTELLECTUAL PROPRIETARY RIGHTS
All information, materials, images, texts, digital files, technologies, software, distinctive signs and other intangible goods contained in the site are protected by copyright, software and other intellectual property laws and are owned by WEG. Any violation of these rights by the user will imply the adoption of applicable legal measures, without prejudice to the immediate termination of this Agreement.
9. THE CONTRIBUTIONS TO THE SITE
The user may send suggestions, comments to the site. By submitting any contribution to this site (including texts, photographs, illustrations, videos, audio files and other materials), the user authorizes the website to use such content free of charge and in no way, by any means, for publication, reproduction, transmission, retransmission, distribution, communication to the public, editing, adaptation and other transformations, translation into any language.
If you do not agree to authorize the site to use your contribution as above, you should not submit any material to the site.
The contributions of the user shall not contain a violation of the copyright or personality of third parties, nor should they indicate abusive, offensive conduct, contrary to good customs, law and public order.
This Terms and Conditions was last updated on September 14, 2018.

redpajama_set_name: RedPajamaCommonCrawl


At the moment you are the director of Ruhrtriennale. Mid-August, your last season starts. What are your priorities for your 2017 program?
JS: We will focus on ideas of a future world. What are our visions for the future, our utopia or dystopia? In this context we will stage the world premiere of a new opera based on "Kein Licht." by Nobel Prize winner Elfriede Jelinek. We will bring two trilogies we started two years ago to their conclusions: a dance performance inspired by Dante's "Divine Comedy" by US choreographer Richard Siegal and an eleven hour theater marathon based on novels by Émile Zola, directed by Luk Perceval. We will show "Pelléas et Mélisande", which is considered one of the first operas of the modern era. Barbara Hannigan stars as Mélisande. We will celebrate electronic pop music with several concerts, featuring Nicolas Jaar, SOHN or Arca. And I myself will direct a music theater adaptation of Don DeLillo's great novel "Cosmopolis". Only to mention a few attractions of our six week festival in the incomparable post-industrial venues that make the Ruhrtriennale so very special.
Afterwards you will be working for Schauspielhaus Bochum. What keeps you in the Ruhr area?
Without revealing too much: what is your vision for Schauspielhaus Bochum?
JS: Schauspielhaus Bochum has a long tradition. We will celebrate the 100th anniversary in our first season 2018/2019. From the early days on it was critically acclaimed and celebrated, especially in the 70s and 80s, when some of the most exciting directors used to work there like Peter Zadek or Claus Peymann. Since then, Schauspielhaus Bochum has been a legend in the German speaking theater world. I wish to build on those great times. I want to create an ensemble of some of the best actresses and actors I know –and also bring some fresh talent to Bochum. Together we want to present strong theater that tells profound stories. And I want to expand the range: with dance, with music concerts, maybe even with video art. Different nationalities and artistic branches shall in influence each other. I wish to create a theater without borders – which is in my opinion just the right thing to do in the current political situation: showing people the richness of our diverse world.
Besides Jahrhunderthalle and Schauspielhaus: what is your favorite place or location in Bochum?
JS: Well, the Schauspielhaus definitely is my favorite place, as I already said. But if I had to mention another place in Bochum besides that, it probably would be a café or a restaurant. There are a few good ones in Bochum where I have already been several times. Good cooking (and eating) is probably my second favorite thing apart from theater.


text:
Carbs Needed to Evolve Big Brains
Neuroscience News August 6, 2015
The importance of dietary carbohydrate in human evolution.
Understanding how and why we evolved such large brains is one of the most puzzling issues in the study of human evolution. It is widely accepted that brain size increase is partly linked to changes in diet over the last 3 million years, and increases in meat consumption and the development of cooking have received particular attention from the scientific community. In a new study published in The Quarterly Review of Biology, Dr. Karen Hardy and her team bring together archaeological, anthropological, genetic, physiological and anatomical data to argue that carbohydrate consumption, particularly in the form of starch, was critical for the accelerated expansion of the human brain over the last million years, and coevolved both with copy number variation of the salivary amylase genes and controlled fire use for cooking.
With global increase in obesity and diet-related metabolic diseases, interest has intensified in ancestral or 'Palaeolithic' diets, not least because – to a first order of approximation – human physiology should be optimized for the nutritional profiles we have experienced during our evolution. Up until now, there has been a heavy focus on the role of animal protein and cooking in the development of the human brain over the last 2 million years, and the importance of carbohydrate, particular in form of starch-rich plant foods, has been largely overlooked.
Hardy's team highlights the following observations to build a case for dietary carbohydrate being essential for the evolution of modern big-brained humans:
(1) The human brain uses up to 25% of the body's energy budget and up to 60% of blood glucose. While synthesis of glucose from other sources is possible, it is not the most efficient way, and these high glucose demands are unlikely to have been met on a low carbohydrate diet;
(2) Human pregnancy and lactation place additional demands on the body's glucose budget and low maternal blood glucose levels compromise the health of both the mother and her offspring;
(3) Starches would have been readily available to ancestral human populations in the form of tubers, as well as in seeds and some fruits and nuts;
(4) While raw starches are often only poorly digested in humans, when cooked they lose their crystalline structure and become far more easily digested;
(5) Salivary amylase genes are usually present in many copies (average ~6) in humans, but in only 2 copies in other primates. This increases the amount of salivary amylase produced and so increases the ability to digest starch. The exact date when salivary amylase genes multiplied remains uncertain, but genetic evidence suggests it was at some point in the last 1 million years.
The human brain uses up to 25% of the body's energy budget and up to 60% of blood glucose. Image is for illustrative purposes only.
Hardy proposes that after cooking became widespread, the co-evolution of cooking and higher copy number of the salivary amylase (and possibly pancreatic amylase) genes increased the availability of pre-formed dietary glucose to the brain and fetus, which in turn, permitted the acceleration in brain size increase which occurred from around 800,000 years ago onwards.
Eating meat may have kick-started the evolution of bigger brains, but cooked starchy foods together with more salivary amylase genes made us smarter still.
About this evolutionary neuroscience and diet research
Source: Karen Hardy – University of Chicago Press Journals
Image Source: The image is credited to Institute of Physics
Original Research: Full open access research for "The Importance of Dietary Carbohydrate in Human Evolution" by Karen Hardy, Jennie Brand-Miller, Katherine D. Brown, Mark G. Thomas, and Les Copeland in The Quarterly Review of Biology. Published online August 2015 doi:10.1086/682587
The Importance of Dietary Carbohydrate in Human Evolution
We propose that plant foods containing high quantities of starch were essential for the evolution of the human phenotype during the Pleistocene. Although previous studies have highlighted a stone tool-mediated shift from primarily plant-based to primarily meat-based diets as critical in the development of the brain and other human traits, we argue that digestible carbohydrates were also necessary to accommodate the increased metabolic demands of a growing brain. Furthermore, we acknowledge the adaptive role cooking played in improving the digestibility and palatability of key carbohydrates. We provide evidence that cooked starch, a source of preformed glucose, greatly increased energy availability to human tissues with high glucose demands, such as the brain, red blood cells, and the developing fetus. We also highlight the auxiliary role copy number variation in the salivary amylase genes may have played in increasing the importance of starch in human evolution following the origins of cooking. Salivary amylases are largely ineffective on raw crystalline starch, but cooking substantially increases both their energy-yielding potential and glycemia. Although uncertainties remain regarding the antiquity of cooking and the origins of salivary amylase gene copy number variation, the hypothesis we present makes a testable prediction that these events are correlated.
"The Importance of Dietary Carbohydrate in Human Evolution" by Karen Hardy, Jennie Brand-Miller, Katherine D. Brown, Mark G. Thomas, and Les Copeland in The Quarterly Review of Biology. Published online August 2015 doi:10.1086/682587
anthropologyblood glucosebraincarbohydratesdietdigestionevolutionevolutionary neuroscienceGeneticslactationNeuroscienceOpen Accessopen sciencepalaeolithic dietpaleo dietproteinsalivary amylase genes
August 7, 2015 Kiumars Lalezarzadeh
Cooked yeast eukaryotes or prokaryotes!? Does this mean beer may have the effect too?
How could this hypothesis support the size of Nephilim brain, where by the giants cooked their carbs? The contrary may be true that telomere size that may correlated with size of organs and body, may shorten due to ethanol or cooked food, or "hot genes". And if alcohol (ethanol) would have resulted by fermentation of starch of carbohydrates, that would predict shortening of telomere, and thus smaller brains. Besides, Glucoamylase can come about by yeast which yet can covert starch into ethonol (http://www.ncbi.nlm.nih.gov/pubmed/18629955)- again posing problem for the starch, or even any yeast hypothesis contributing to brain evolution. One may suppose that yeast nucleotides may help extent neurites because of hypoxia caused by yeast, but not the neuronal cell body containing the telomere. Would it be more likely that the human brain may had even reduced in size, or injured by starch and carbohydrates due to alcohol or genetic ultra structure of wheat (Triticum), for example? if so, could be that brain contours may become more elaborate and size reduced due to cooked starch or alcohol?


text:
>Richard H. Speier
Richard H. Speier
Adjunct Staff
Off Site Office
B.A. in physics, Harvard University; Ph.D. in political science, MIT
One Page Bio
Richard Speier is Adjunct Staff with the RAND Corp. He received a Bachelor of Arts degree in Physics from Harvard College and a Ph.D. in Political Science from the Massachusetts Institute of Technology. During 4 years of graduate study he focused on economic and strategic analysis, including work at the RAND Corporation and the Institute for Defense Analysis.
He then entered the Office of Management and Budget, where he helped reshape nuclear and space programs, and later the Arms Control and Disarmament Agency, where he analyzed nuclear technologies that contribute to proliferation. He spent a year at the National War College co-authoring a prize-winning study, "The Bomb in Southwest Asia".
In 1982 he joined the Office of the Secretary of Defense to start the Office of Non- Proliferation Policy. For 10 years he helped design, negotiate, and implement the Missile Technology Control Regime. In recognition of his work on the international negotiations that led to the MTCR, the Secretary of Defense awarded him the Meritorious Civilian Service Medal. Speier is widely credited with being "the father of the MTCR." He later helped initiate the Defense Department's Office of Counter-Proliferation Policy.
Speier now consults in the Washington, D.C. area. He has worked with both non-profit and for-profit organizations. His dozens of writings include a history of the negotiation of the MTCR, a study of nonproliferation sanctions, and articles on the implications of unmanned air vehicles, space launch vehicles, missile defense, and hypersonic missiles.
Hypersonic Missiles: A New Proliferation Challenge
Within 10 years, hypersonic missiles are likely to be deployed and offered on the international market. But there is time for action by states that do not want hypersonic missiles to flourish in their neighborhoods. It is time to move toward heading off this threat while it is still possible to do so.
Mar 29, 2018 Georgetown Journal of International Affairs
Hypersonic Missile Nonproliferation: Hindering the Spread of a New Class of Weapons
This report examines the implications of the proliferation of hypersonic missiles and possible measures to hinder it.
Cruise Missile Penaid Nonproliferation: Hindering the Spread of Countermeasures Against Cruise Missile Defenses
An attacker's missile-borne countermeasures to cruise missile defenses are known as penetration aids, or penaids. This research recommends export controls on penaid-related items under the Missile Technology Control Regime.
Penaid Nonproliferation: Hindering the Spread of Countermeasures Against Ballistic Missile Defenses
An attacker's missile-borne countermeasures to ballistic missile defenses are known as penetration aids, or penaids. This research recommends export controls on penaid-related items under the Missile Technology Control Regime.
RAND Review: Vol. 26, No. 1, Spring 2002
Suggests that the new national agenda of high-stakes testing in K-12 schools may be more of an academic hindrance than a help; also discusses ways to take the profit out of WMD proliferation, U.S. Army logistics, and the 2002 general election.
Nonproliferation Sanctions
This study examines the United States' use of sanctions against foreign entities to prevent the proliferation of nuclear, biological, and chemical weapons.
Asiatom: Proposals, Alternatives and Next Steps
Japan and other East Asian countries have recently proposed an Asiatom, similar to Euratom in western Europe, for joint efforts in various nuclear programs. The authors examine this proposal and its implications for nuclear proliferation.
A Concept for Strategic Material Accelerated Removal Talks (SMART)
The United States, Russia, and other countries should ensure that weapon-usable material is not subject to national diversion. The quickest way to achieve this goal is to transport excess weapon-usable material to a secure foreign storage site.
The Proposed Fissile-Material Production Cutoff: Next Steps
Examines and proposes some "next steps" to the proposed convention, to further strengthen worldwide control of weapon-usable material.
Hypersonic Missile Nonproliferation
Hypersonic missile proliferation would increase the chances of strategic war. It would give nations an incentive to become trigger-happy.
Source: RAND News Release
Related RAND Resources
All Policy Experts
Public Policy Experts Guide


The shots that fired the bullets that entered Pumpkin's head were heard by no less than eight people. Three instinctively closed their windows, checked their door locks, and withdrew to the safety, or at least the seclusion, of their small apartments. Two others, each with experience in such matters, ran from the vicinity as fast if not faster than the gunman himself. Another, the neighborhood recycling fanatic, was digging through some garbage in search of aluminum cans when he heard the sharp sounds of the daily skirmish, very nearby. He jumped behind a pile of cardboard boxes until the shelling stopped, then eased into the alley where he saw what was left of Pumpkin.
And two saw almost everything. They were sitting on plastic milk crates, at the corner of Georgia and Lamont in front of a liquor store, partially hidden by a parked car so that the gunman, who glanced around briefly before following Pumpkin into the alley, didn't see them. Both would tell the police that they saw the boy with the gun reach into his pocket and pull it out; they saw the gun for sure, a small black pistol. A second later they heard the shots, though they did not actually see Pumpkin take them in the head. Another second and the boy with the gun darted from the alley and, for some reason, ran straight in their direction. He ran bent at the waist, like a scared dog, guilty as hell. He wore red-and-yellow basketball shoes that seemed five sizes too big and slapped the pavement as he made his getaway.
When he ran by them he was still holding the gun, probably a.38, and he flinched just for an instant when he saw them and realized they had seen too much. For one terrifying second, he seemed to raise the gun as if to eliminate the witnesses, both of whom managed to flip backward from their plastic milk crates and scramble off in a mad flurry of arms and legs. Then he was gone.
One of them opened the door to the liquor store and yelled for someone to call the police, there had been a shooting.
Thirty minutes later, the police received a call that a young man matching the description of the one who had wasted Pumpkin had been seen twice on Ninth Street carrying a gun in open view and acting stranger than most of the people on Ninth. He had tried to lure at least one person into an abandoned lot, but the intended victim had escaped and reported the incident.
The police found their man an hour later. His name was Tequila Watson, black male, age twenty, with the usual drug-related police record. No family to speak of. No address. The last place he'd been sleeping was a rehab unit on W Street. He'd managed to ditch the gun somewhere, and if he'd robbed Pumpkin then he'd also thrown away the cash or drugs or whatever the booty was. His pockets were clean, as were his eyes. The cops were certain Tequila was not under the influence of anything when he was arrested. A quick and rough interrogation took place on the street, then he was handcuffed and shoved into the rear seat of a D.C. police car.
They drove him back to Lamont Street, where they arranged an impromptu encounter with the two witnesses. Tequila was led into the alley where he'd left Pumpkin. "Ever been here before?" a cop asked.
Tequila said nothing, just gawked at the puddle of fresh blood on the dirty concrete. The two witnesses were eased into the alley, then led quietly to a spot near Tequila.
"That's him," both said at the same time.
"He's wearing the same clothes, same basketball shoes, everything but the gun."
Tequila was shoved into the car once again and taken to jail. He was booked for murder and locked away with no immediate chance of bail. Whether through experience or just fear, Tequila never said a word to the cops as they pried and cajoled and even threatened. Nothing incriminating, nothing helpful. No indication of why he would murder Pumpkin. No clue as to their history, if one existed at all. A veteran detective made a brief note in the file that the killing appeared a bit more random than was customary.
No phone call was requested. No mention of a lawyer or a bail bondsman. Tequila seemed dazed but content to sit in a crowded cell and stare at the floor.
Pumpkin had no traceable father but his mother worked as a security guard in the basement of a large office building on New York Avenue. It took three hours for the police to determine her son's real name - Ramon Pumphrey - to locate his address, and to find a neighbor willing to tell them if he had a mother.
Adelfa Pumphrey was sitting behind a desk just inside the basement entrance, supposedly watching a bank of monitors. She was a large thick woman in a tight khaki uniform, a gun on her waist, a look of complete disinterest on her face. The cops who approached her had done so a hundred times. They broke the news, then found her supervisor.
In a city where young people killed each other every day, the slaughter had thickened skins and hardened hearts, and every mother knew many others who'd lost their children. Each loss brought death a step closer, and every mother knew that any day could be the last. The mothers had watched the others survive the horror. As Adelfa Pumphrey sat at her desk with her face in her hands, she thought of her son and his lifeless body lying somewhere in the city at that moment, being inspected by strangers.
She swore revenge on whoever killed him.
She cursed his father for abandoning the child.
She cried for her baby.
And she knew she would survive. Somehow, she would survive.
Adelfa went to court to watch the arraignment. The police told her the punk who'd killed her son was scheduled to make his first appearance, a quick and routine matter in which he would plead not guilty and ask for a lawyer. She was in the back row with her brother on one side and a neighbor on the other, her eyes leaking tears into a damp handkerchief. She wanted to see the boy. She also wanted to ask him why, but she knew she would never get the chance.
They herded the criminals through like cattle at an auction. All were black, all wore orange coveralls and handcuffs, all were young. Such waste.
In addition to his handcuffs, Tequila was adorned with wrist and ankle chains since his crime was especially violent, though he looked fairly harmless when he was shuffled into the courtroom with the next wave of offenders. He glanced around quickly at the crowd to see if he recognized anyone, to see if just maybe someone was out there for him. He was seated in a row of chairs, and for good measure one of the armed bailiffs leaned down and said, "That boy you killed. That's his mother back there in the blue dress."
With his head low, Tequila slowly turned and looked directly into the wet and puffy eyes of Pumpkin's mother, but only for a second. Adelfa stared at the skinny boy in the oversized coveralls and wondered where his mother was and how she'd raised him and if he had a father, and, most important, how and why his path had crossed that of her boy's. The two were about the same age as the rest of them, late teens or early twenties. The cops had told her that it appeared, at least initially, that drugs were not involved in the killing. But she knew better. Drugs were involved in every layer of street life. Adelfa knew it all too well. Pumpkin had used pot and crack and he'd been arrested once, for simple possession, but he had never been violent. The cops were saying it looked like a random killing. All street killings were random, her brother had said, but they all had a reason.
On one side of the courtroom was a table around which the authorities gathered. The cops whispered to the prosecutors, who flipped through files and reports and tried valiantly to keep the paperwork ahead of the criminals. On the other side was a table where the defense lawyers came and went as the assembly line sputtered along. Drug charges were rattled off by the Judge, an armed robbery, some vague sexual attack, more drugs, lots of parole violations. When their names were called, the defendants were led forward to the bench, where they stood in silence. Paperwork was shuffled, then they were hauled off again, back to jail.


Black Jack, 21, Texas HoldEm, Roulette. Nice drinks, elegant ladies, quality service and everything you need from a real Casino. A splendid place which meets all your needs if you are ready to gamble. A place for gentlemen like you. Don't forget to win as it's easier to drink afterwards until dawn.


text:
Q: web scraping get content from xml how can i use php to get the contents from a xml page.
the content as follows:
 <entry>
   <title>News</title>
     <link rel="alternate" href="http://www.website.com/detail/2688327:BlogPost:1569917"/>
     <id>tag:www.website.com,2012-04-25:2688327:BlogPost:1569917</id>
     <updated>2012-04-25T08:30:00.000Z</updated>
     <author>
     <name>Username</name>
     <uri>http://www.website.com/profile/username</uri>
     </author>
      <summary type="html">
      Hi this is the latest news
      </summary>
</entry>

 <entry>
   <title>News2</title>
     <link rel="alternate" href="http://www.website.com/detail/2688327:BlogPost:1569917"/>
     <id>tag:www.website.com,2012-04-25:2688327:BlogPost:1569917</id>
     <updated>2012-04-25T08:30:00.000Z</updated>
     <author>
     <name>Username2</name>
     <uri>http://www.website.com/profile/username</uri>
     </author>
      <summary type="html">
      Hi this is the latest news
      </summary>
</entry>

 <entry>
   <title>News3</title>
     <link rel="alternate" href="http://www.website.com/detail/2688327:BlogPost:1569917"/>
     <id>tag:www.website.com,2012-04-25:2688327:BlogPost:1569917</id>
     <updated>2012-04-25T08:30:00.000Z</updated>
     <author>
     <name>Username3</name>
     <uri>http://www.website.com/profile/username</uri>
     </author>
      <summary type="html">
      Hi this is the latest news
      </summary>
</entry>

 <entry>
   <title>News4</title>
     <link rel="alternate" href="http://www.website.com/detail/2688327:BlogPost:1569917"/>
     <id>tag:www.website.com,2012-04-25:2688327:BlogPost:1569917</id>
     <updated>2012-04-25T08:30:00.000Z</updated>
     <author>
     <name>Username4</name>
     <uri>http://www.website.com/profile/username</uri>
     </author>
      <summary type="html">
      Hi this is the latest news
      </summary>
</entry>

How can i get an array of the title , the blog link <link rel="alternate" href="http://www.website.com/detail/2688327:BlogPost:1569917"/> , the author details like name and uri(profile link) and the summary out using php?

A: check out simplexml, xpath
http://php.net/manual/en/book.simplexml.php
   $file = 'url or file name';
    $xml = simplexml_load_file('$file');
    $list= $xml->xpath("/entry"); // root/entry ...
    print $list[0]->id;
    #var_dump($list);

redpajama_set_name: RedPajamaStackExchange


Прошу Вас принять меры к владельцам и пользователи аккаунтов в различных социальных интернет-сетях (русскоязычные сегменты Twitter и Facebook), активно содействовавшим в организации массовых беспорядков 6 мая 2012 года в г. Москве.
I ask you to take measures against the owners and users of accounts on various online social networks ([such as] the Russian language segments of Twitter and Facebook), who actively administered the organization of mass unrest on May 6, 2012, in Moscow.
Khinshtein added that bloggers are also using these services to advocate violence against "the legally elected President" and Russia's "constitutional order." (Twitter user Lucius Aevus, for instance, wrote [ru] on April 24, "Buy weapons, stockpile food, join up, and kill Putin.") Khinshtein called specifically for (1) the prevention of the further spread of such illegal, extremist online appeals, (2) the criminal prosecution of individuals who use social networks to disseminate extremism, and (3) any other measures legally necessary.
Screenshot of Khinshtein's official Duma webpage, 15 May 2012.
Hours after news of this petition caused a stir in the media, Khinshtein responded by publishing the full text of his letter to Chaika, claiming [ru] that his initiative is not a plea for political censorship. Khinshtein did not, however, publish the supplementary materials he attached to the letter, where he cataloged dozens of examples of bloggers advocating illegal activity. The list, to which online newspaper Gazeta.ru has access and has graciously summarized [ru], focuses on nationalist and liberal democratic oppositionists — some anonymous and others well-known.
First on the list is self-described "ultra-right" anonymous blogger odinzavseh ("one for all"), whom Khinshtein links to soccer hooligans, like the ones infamous for inciting a small ethnic riot in Moscow in December 2010. On Twitter, odinzavseh has called on protesters to beat up Putin-supporters and public figures Sergei Minaev and Konstantin Rykov. He also allegedly encouraged oppositionists to attack government buildings with Molotov cocktails on the night of May 8. Both these tweets were later deleted, though evidence of the former can still be found [ru] on Yandex. In what seems particularly outrageous to many bloggers, Khinshtein has claimed, as well, that retweeting (reposting another user's message) also carries criminal liability. Odinzavseh, for instance, retweeted user LailaMooore's call [ru] on protesters to slash the tires of police vans and bring weapons to protests.
Не выбрасывайте омоновские шлемы в реку, надевайте их на себя. Это броня, она вас защитит от полицейских дубинок.
Don't throw the police helmets into the river: wear them yourselves. They're armor, and will protect you from their nightsticks.
Отнимайте у ОМОНа щиты, каски, дубинки, это уравнивает ваши шансы. Не нападайте на них, но отнимайте амуницию.
Take away the police's shields, helmets, nightsticks — this evens your chances. Don't attack them, but take away their ammunition.
Думаю, частью стратегии должна стать дальнейшая радикализация масс, чтобы перейти к тактике ощутимой городской герильи.
I think that further radicalization of the masses should become a part [of the opposition], in order to transition to urban guerilla tactics.
Friends, you're all having a laugh, but don't forget that for EVERY RETWEET Khinshtein will question you in the interrogation room of the prison nearest to your home.
Чиновники из мэрии вышли на связь, обещают в течение часа согласовать марш. Если обманут – сожжем мэрию.
Officials from the Mayor's office have been in contact. They promise to approve the [May 6] march within an hour. If they're lying, we'll burn down city hall.
Some RuNet voices have insisted that Khinshtein is asking the Prosecutor General to do something fundamentally beyond its legal authority, and is only causing a stir to grab a moment's publicity. In a series of identical tweets addressed to prominent RuNet bloggers, Yuri Suetin posted the scan [ru] of a letter from May 2011, where the FSB informs a certain Dmitri Goriugin that Russian law enforcement agencies are incapable of deleting or banning posts from Twitter. (Goriugin had apparently complained about personal insults directed at him, published online by pro-Kremlin blogger fritzmorgen — an account that ironically Twitter has since suspended.) Other users, however, were quick to point out that Khinshtein is citing anti-extremism laws that carry far more extensive police powers than Criminal Code 130 (to which Goriugin appealed).
Russian politician, Yuri Chaika, 8 December 2009, photo by Russian Presidential Press and Information Office, CC BY-SA 3.0; Wikimedia Commons.
The state's response to Khinshtein's letter will be an interesting test of the lengths to which the Kremlin is prepared to pressure oppositionists in their use of online social media. Services like Twitter and Facebook are widely credited with energizing the protest movement — indeed, the feasibility of recent mass demonstrations seems to have required these instruments. Likewise, heightened public scrutiny on the extremist content of oppositionists' blog posts and tweets will test the maturity and accountability of a space that until now has hosted language often violent and legally dubious according to Russian legislation.
Interesting article, as always. Thank you for a detailed review.