Rolls back and readds the proxy subscriptions that may have been removed.

@param subMessage
482005
void handleRollback(SubscriptionMessage subMessage, 
                      LocalTransaction transaction)
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "handleRollback", new Object[] { subMessage, transaction });

    try
    {
      if (transaction != null)
      {
        try
        {
          transaction.rollback();
        }
        catch (SIException e)
        {
          // If there are any exceptions, then this is bad, log a FFDC
          FFDCFilter.processException(
            e,
            "com.ibm.ws.sib.processor.proxyhandler.NeighbourProxyListener.handleRollback",
            "1:330:1.73",
            this);
          
          SibTr.exception(tc, e);
        }
      }
    
      Transaction msTran = iProxyHandler.getMessageProcessor().resolveAndEnlistMsgStoreTransaction(transaction);
      
      switch (subMessage.getSubscriptionMessageType().toInt())
      {
        case SubscriptionMessageType.CREATE_INT :
    
          // Remove any created Proxy subscriptions
         iProxyHandler.remoteUnsubscribeEvent(iAddTopicSpaces,
                                              iAddTopics, 
                                              subMessage.getBus(), 
                                              msTran,
                                              false);
    
          break;
    
        case SubscriptionMessageType.DELETE_INT :
    
        // Create any removed Proxy subscriptions
        iProxyHandler.remoteSubscribeEvent(iDeleteTopicSpaces,
                                           iDeleteTopics,                                          
                                           subMessage.getBus(),
                                           msTran, 
                                           false);              
    
          break;
    
        case SubscriptionMessageType.RESET_INT :
    
        // Remove any created Proxy subscriptions
        iProxyHandler.remoteUnsubscribeEvent(iAddTopicSpaces,
                                             iAddTopics,                                            
                                             subMessage.getBus(), 
                                             msTran,
                                             false);
    
        // Create any removed Proxy subscriptions
        iProxyHandler.remoteSubscribeEvent(iDeleteTopicSpaces,
                                           iDeleteTopics,                                            
                                           subMessage.getBus(),
                                           msTran, 
                                           false);
                                                         
          break;
    
        default :
          // Log Message
          break;
      }
    }
    catch (SIException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.proxyhandler.NeighbourProxyListener.handleRollback",
        "1:392:1.73",
        this);
    }
    
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "handleRollback");
  }
601278
public void rollback()
    {
        //  Remove incomplete message from the outbound pipe.
        Msg msg;
        if (outpipe != null) {
            while ((msg = outpipe.unwrite()) != null) {
                assert (msg.hasMore());
            }
        }
    }
262106
@Override
	public void registerConsumer(LocalMessageConsumer consumer)
    {
        super.registerConsumer(consumer);
        
        subscriptionsLock.writeLock().lock();
        try
		{
        	LocalTopicSubscription subscription = subscriptionMap.remove(consumer.getSubscriberId());
        	if (subscription == null)
        	{
        		// New subscription
        		storeNewSubscription(consumer);
        	}
        	else
        	{
        		// Remove old subscription
        		removeSubscription(subscription);
        		// Replace by new subscription
        		storeNewSubscription(consumer);
        	}
		}
        finally
        {
        	subscriptionsLock.writeLock().unlock();
        }
    }
1064404
public void setSubscriptions(java.util.Collection<Subscription> subscriptions) {
        if (subscriptions == null) {
            this.subscriptions = null;
            return;
        }

        this.subscriptions = new java.util.ArrayList<Subscription>(subscriptions);
    }
301618
protected void subAppend(LoggingEvent event) {
		long n = event.timeStamp;
		if (n >= nextCheck) {
			try {
				now.setTime(n);
				cleanupAndRollOver();
			} catch (IOException ioe) {
				if (ioe instanceof InterruptedIOException) {
					Thread.currentThread().interrupt();
				}
				LogLog.error("rollOver() failed.", ioe);
			}
		}
		super.subAppend(event);
	}
1190270
protected static void handleDeleteDurable(
    DestinationManager DM,
    ControlDeleteDurable msg,
    MessageProcessor MP)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "handleDeleteDurable", new Object[] {DM, msg, MP});

    int status = STATUS_SUB_GENERAL_ERROR;
    
    try 
    {
      // Track down a handler for this durable sub, error if we can't find one
      String  subName     = msg.getDurableSubName();
      HashMap durableSubs = DM.getDurableSubscriptionsTable();

      synchronized (durableSubs)
      {
        //Look up the consumer dispatcher for this subId in the system durable subs list
        ConsumerDispatcher cd =
          (ConsumerDispatcher) durableSubs.get(subName);
          
        // Check that the durable subscription existed
        if (cd == null) 
        {
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
            SibTr.exit(tc, "handleDeleteDurable", "SIDurableSubscriptionNotFoundException");
          
          status = DurableConstants.STATUS_SUB_NOT_FOUND;
          
          throw new SIDurableSubscriptionNotFoundException(
            nls.getFormattedMessage(
              "SUBSCRIPTION_DOESNT_EXIST_ERROR_CWSIP0072",
              new Object[] { subName,                
                             MP.getMessagingEngineName() },
              null));
        }
          
        // Obtain the destination from the consumer dispatcher
        // A cast error here will result in an error sent back to the caller
        BaseDestinationHandler handler = cd.getDestination();

        boolean userMatch = true;
        // If security is disabled then we'll bypass the check
        if(MP.isBusSecure())
        {                
          // Check that the user who is attempting to delete this durable subscription
          // matches that set in the CD state when the subscription was created
        
          ConsumerDispatcherState subState = cd.getConsumerDispatcherState();
          String theUser = msg.getSecurityUserid(); 
          if(theUser == null)
          {
            if(subState.getUser() != null)
              userMatch = false;
          }
          else
          {
            if (!theUser.equals(subState.getUser()))
              userMatch = false;  
          }
        }  
        // Handle the case where the users do not match
        if (!userMatch)
        {  
          status = DurableConstants.STATUS_NOT_AUTH_ERROR;
        }        
        else
        {
          status = handler.deleteDurableFromRemote(subName);
        }
      }
    }
    catch (SIDurableSubscriptionNotFoundException e) 
    {
       // No FFDC code needed
       SibTr.exception(tc, e);
    }
    catch (Exception e)
    {
      // FFDC
      // Any other exception is unexpected
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.DurableOutputHandler.handleDeleteDurable",
        "1:387:1.45.1.1",
        DurableOutputHandler.class);
    }
    
    // Forward the status to the caller and we're done
    SIBUuid8              sender = msg.getGuaranteedSourceMessagingEngineUUID(); 
    ControlDurableConfirm reply  = createDurableConfirm(MP, sender, msg.getRequestID(), status);
    MP.getMPIO().sendToMe(sender, SIMPConstants.CONTROL_MESSAGE_PRIORITY, reply);

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "handleDeleteDurable");
  }
1295190
protected void recoverSubscriptions(MultiMEProxyHandler proxyHandler) throws MessageStoreException, SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "recoverSubscriptions", proxyHandler);

    iProxyHandler = proxyHandler;

    iProxies = new Hashtable();

    NonLockingCursor cursor = null;
    try
    {
      cursor = newNonLockingItemCursor(new ClassEqualsFilter(MESubscription.class));

      AbstractItem item = null;
      while ((item = cursor.next()) != null)
      {
        MESubscription sub = null;
          sub = (MESubscription)item;

        // Generate a key for the topicSpace/topic
        final String key = BusGroup.subscriptionKey(sub.getTopicSpaceUuid(),
                                                       sub.getTopic());

        // Add the Neighbour into the list of recovered Neighbours
        iProxies.put(key, sub);

        // Having created the Proxy, need to readd the proxy to the matchspace or just reference it.
        iNeighbours.createProxy(this,
                                iDestinationManager.getDestinationInternal(sub.getTopicSpaceUuid(), false),
                                sub,
                                sub.getTopicSpaceUuid(),
                                sub.getTopic(),
                                true);

        // When recovering subscriptions, we need to call the event post commit
        // code to either add a reference, or put in the MatchSpace.
        sub.eventPostCommitAdd(null);

      }
    }
    finally
    {
      if (cursor != null)
        cursor.finished();
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "recoverSubscriptions");
  }
537039
protected synchronized void unmapSubscrip (int oid)
    {
        ClientProxy rec;
        synchronized (_subscrips) {
            rec = _subscrips.remove(oid);
        }
        if (rec != null) {
            rec.unsubscribe();
        } else {
            boolean alreadyDestroyed = _destroyedSubs.remove(oid);
            if (!alreadyDestroyed) {
                log.warning("Missing subscription in unmap", "client", this, "oid", oid);
            }
        }
    }
1263786
public void processReceivedMessage( Message message, Session session ) throws IOException {
		this.logger.fine( "The DM's HTTP client is about to process a message (" + message + ") received through a web socket." );

		// HttpMessage
		if( message instanceof HttpMessage ) {
			HttpMessage httpMsg = (HttpMessage) message;

			// Store the session
			registerSession( httpMsg.getOwnerId(), session );

			// Publish the message
			publish( httpMsg.getCtx(), httpMsg.getMessage());
		}

		// Subscription message
		else if( message instanceof SubscriptionMessage ) {
			SubscriptionMessage sub = (SubscriptionMessage) message;

			// Store the session
			registerSession( sub.getOwnerId(), session );

			// Update the subscriptions
			if( sub.isSubscribe())
				subscribe( sub.getOwnerId(), sub.getCtx());
			else
				unsubscribe( sub.getOwnerId(), sub.getCtx());
		}
	}
1254738
protected void subscriptionResponseReceived(IoSession session,
			SubscriptionMessage subscriptionMessage) {
		log.info("Received {}", subscriptionMessage);
		this.receivedSubscription = subscriptionMessage;

		if (this.sentSubscription == null) {
			log.error(
					"Protocol error: Received a subscription response without sending a request.\n{}",
					subscriptionMessage);
			this._disconnect();
			return;
		}

		if (!this.sentSubscription.equals(this.receivedSubscription)) {
			log.info(
					"Server did not fully accept subscription request.\nOriginal:\n{}\nAmended\n{}",
					this.sentSubscription, this.receivedSubscription);
		}

		for (ConnectionListener listener : this.connectionListeners) {
			listener.subscriptionReceived(this, subscriptionMessage);
		}
	}
117654
private void removeOverlappingSubscriptionEvents(final LinkedList<SubscriptionEvent> events) {
        final Iterator<SubscriptionEvent> iterator = events.iterator();
        final Map<String, DefaultSubscriptionEvent> prevPerService = new HashMap<String, DefaultSubscriptionEvent>();
        while (iterator.hasNext()) {
            final DefaultSubscriptionEvent current = (DefaultSubscriptionEvent) iterator.next();
            final DefaultSubscriptionEvent prev = prevPerService.get(current.getServiceName());
            if (prev != null) {
                if (current.overlaps(prev)) {
                    iterator.remove();
                } else {
                    prevPerService.put(current.getServiceName(), current);
                }
            } else {
                prevPerService.put(current.getServiceName(), current);
            }
        }
    }
675872
@Override
	public void postProcessBeforeDestruction(final Object bean, final String beanName) throws BeansException {
		if (subscribers.containsKey(beanName)) {
			Subscribe annotation = getAnnotation(bean.getClass(), beanName);
			LOG.debug("Unsubscribing the event listener '{}' from event bus '{}' with topic '{}", beanName,
					annotation.eventBus(), annotation.topic());
			try {
				EventBus eventBus = getEventBus(annotation.eventBus(), beanName);
				eventBus.unsubscribe(annotation.topic(), (EventSubscriber) bean);
			} catch (Exception e) {
				LOG.error("Unsubscribing the event listener '{}' failed", beanName, e);
			} finally {
				subscribers.remove(beanName);
			}
		}
	}
94973
private void decodeSubscription(ByteBuf in, SubscribeMessage message) throws UnsupportedEncodingException {
        String topic = Utils.decodeString(in);
        byte qos = (byte)(in.readByte() & 0x03);
        message.addSubscription(new SubscribeMessage.Couple(qos, topic));
    }
1168994
private void terminateSubscriptions(IoSession session, HeapState state, String heapUri, Subscription.CloseReason reason) {
        Lock heapUpdateLock = null;
        try {
            if (state != null) {
                heapUpdateLock = state.getUpdateLock();
                heapUpdateLock.lock();
            }
            subTermLock.lock();

            if (state != null) {
                if (!state.isTerminated()) {
                    state.terminateSubscriptions(session, reason);
                    if (state.getSessions().isEmpty()) {
                        terminateSubscriptions(heapUri, reason);
                    }
                }
            }

            Multiset<String> heapsForSession = heapsByClient.get(session);
            if (heapsForSession != null) {
                nioLogger.log(NioLogger.LoggingLevel.TRANSPORT, session, "Terminating subscription on %s heaps", heapsForSession.keySet().size());
                heapsForSession.removeAll(heapUri);
                if (heapsForSession.isEmpty()) {
                    terminateSubscriptions(session, reason);
                }
            }
        } finally {
            subTermLock.unlock();
            if (heapUpdateLock != null) {
                heapUpdateLock.unlock();
            }
        }
    }
356854
public void assertSubscriptions() {
        for (String channel : subscribers.keySet()) {
            try {
                subscribers.put(channel, null);
                subscribe(channel);
            } catch (Throwable e) {
                break;
            }
        }
    }
307921
protected void resetReplySubscriptionMessage()
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "resetReplySubscriptionMessage");

    // Reset the state
    reset();
    
    // Indicate that this is a create message    
    iSubscriptionMessage.setSubscriptionMessageType(
      SubscriptionMessageType.REPLY);

    if (tc.isEntryEnabled())
      SibTr.exit(tc, "resetReplySubscriptionMessage");
  }
472813
@Override
    public void setSubscriptionDurability(final String subscriptionDurability) {
        if (TraceComponent.isAnyTracingEnabled() && TRACE.isDebugEnabled()) {
            SibTr.debug(this, TRACE, "setSubscriptionDurability",
                        subscriptionDurability);
        }
        _subscriptionDurability = subscriptionDurability;
    }