/*
(non-Javadoc)

@see com.ibm.ws.sib.processor.MPConsumerSession#registerStoppableAsynchConsumerCallback()
773987
@Override
    public void registerStoppableAsynchConsumerCallback(
                                                        StoppableAsynchConsumerCallback callback,
                                                        int maxActiveMessages,
                                                        long messageLockExpiry,
                                                        int maxBatchSize,
                                                        Reliability unrecoverableReliability,
                                                        boolean inLine,
                                                        OrderingContext extendedMessageOrderingContext,
                                                        int maxSequentialFailures,
                                                        long hiddenMessageDelay)
                    throws SISessionUnavailableException, SISessionDroppedException, SIIncorrectCallException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "registerStoppableAsynchConsumerCallback",
                        new Object[] { callback,
                                      maxActiveMessages,
                                      messageLockExpiry,
                                      maxBatchSize,
                                      unrecoverableReliability,
                                      inLine,
                                      extendedMessageOrderingContext,
                                      maxSequentialFailures,
                                      hiddenMessageDelay });

        //register the AsynchConsumer with the LCP
        _localConsumerPoint.registerStoppableAsynchConsumer(
                                                            callback,
                                                            maxActiveMessages,
                                                            messageLockExpiry,
                                                            maxBatchSize,
                                                            unrecoverableReliability,
                                                            inLine,
                                                            (OrderingContextImpl) extendedMessageOrderingContext,
                                                            null, // No callback busy lock is used for non-mediation consumers.
                                                            maxSequentialFailures,
                                                            hiddenMessageDelay);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "registerStoppableAsynchConsumerCallback");
    }
235494
private void registerGlobalTxTable(GlobalComponentRegistry globalComponentRegistry) {
      InternalCacheRegistry registry = globalComponentRegistry.getComponent(InternalCacheRegistry.class);
      ConfigurationBuilder builder = new ConfigurationBuilder();
      //we can't lose transactions. distributed cache can lose data is num_owner nodes crash at the same time
      builder.clustering().cacheMode(globalComponentRegistry.getGlobalConfiguration().isClustered() ?
            CacheMode.REPL_SYNC :
            CacheMode.LOCAL);
      builder.transaction().transactionMode(TransactionMode.NON_TRANSACTIONAL);
      //persistent? should we keep the transaction after restart?
      registry.registerInternalCache(GLOBAL_TX_TABLE_CACHE_NAME, builder.build(),
            EnumSet.noneOf(InternalCacheRegistry.Flag.class));
   }
1368834
private ServiceRegistration<?> registerMBeanService(String jmsResourceName, BundleContext bundleContext) {
        Dictionary<String, String> props = new Hashtable<String, String>();
        props.put(KEY_SERVICE_VENDOR, "IBM");

        JmsServiceProviderMBeanImpl jmsProviderMBean = new JmsServiceProviderMBeanImpl(getServerName(), KEY_JMS2_PROVIDER);
        props.put(KEY_JMX_OBJECTNAME, jmsProviderMBean.getobjectName());

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "JmsQueueMBeanImpl=" + jmsProviderMBean.getobjectName() + " props=" + props);
        }
        return bundleContext.registerService(JmsServiceProviderMBeanImpl.class, jmsProviderMBean, props);
    }
624435
public void dispatchConsumerSetChangeCallbackEvent(ConsumerSetChangeCallback consumerSetChangeCallback,boolean isEmpty)
   {
      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "dispatchConsumerSetChangeCallbackEvent",
            new Object[]{consumerSetChangeCallback, isEmpty});

      //Create a new ConsumerSetChangeCallbackThread and dispatch it.
      final ConsumerSetChangeCallbackThread thread = new ConsumerSetChangeCallbackThread(consumerSetChangeCallback,isEmpty);
      dispatchThread(thread);

      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "dispatchConsumerSetChangeCallbackEvent");
   }
965173
@Override
    public void registerAsynchConsumerCallback(
                                               AsynchConsumerCallback callback,
                                               int maxActiveMessages,
                                               long messageLockExpiry,
                                               int maxBatchSize,
                                               OrderingContext context)
                    throws SISessionUnavailableException, SISessionDroppedException,
                    SIErrorException,
                    SIIncorrectCallException
    {
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.entry(CoreSPIConsumerSession.tc, "registerAsynchConsumerCallback",
                        new Object[] { this,
                                      callback,
                                      new Integer(maxActiveMessages),
                                      new Long(messageLockExpiry),
                                      new Integer(maxBatchSize),
                                      context });
        registerAsynchConsumerCallback(callback,
                                       maxActiveMessages,
                                       messageLockExpiry,
                                       maxBatchSize,
                                       null,
                                       false,
                                       context);

        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.exit(CoreSPIConsumerSession.tc, "registerAsynchConsumerCallback");
    }
1004732
public void connectAsynch(Object address, ConnectionReadyCallback inputAppCallback) throws IllegalStateException {
        this.appCallback = inputAppCallback;
        this.appSideChannelLink.setApplicationCallback(this);
        this.appSideChannelLink.connectAsynch(address);
    }
507776
@SuppressWarnings("unchecked")
	private void register(final G gateway, final int attempt, final long timeoutMillis) {
		// eager check for canceling to avoid some unnecessary work
		if (canceled) {
			return;
		}

		try {
			log.info("Registration at {} attempt {} (timeout={}ms)", targetName, attempt, timeoutMillis);
			CompletableFuture<RegistrationResponse> registrationFuture = invokeRegistration(gateway, fencingToken, timeoutMillis);

			// if the registration was successful, let the TaskExecutor know
			CompletableFuture<Void> registrationAcceptFuture = registrationFuture.thenAcceptAsync(
				(RegistrationResponse result) -> {
					if (!isCanceled()) {
						if (result instanceof RegistrationResponse.Success) {
							// registration successful!
							S success = (S) result;
							completionFuture.complete(Tuple2.of(gateway, success));
						}
						else {
							// registration refused or unknown
							if (result instanceof RegistrationResponse.Decline) {
								RegistrationResponse.Decline decline = (RegistrationResponse.Decline) result;
								log.info("Registration at {} was declined: {}", targetName, decline.getReason());
							} else {
								log.error("Received unknown response to registration attempt: {}", result);
							}

							log.info("Pausing and re-attempting registration in {} ms", retryingRegistrationConfiguration.getRefusedDelayMillis());
							registerLater(gateway, 1, retryingRegistrationConfiguration.getInitialRegistrationTimeoutMillis(), retryingRegistrationConfiguration.getRefusedDelayMillis());
						}
					}
				},
				rpcService.getExecutor());

			// upon failure, retry
			registrationAcceptFuture.whenCompleteAsync(
				(Void v, Throwable failure) -> {
					if (failure != null && !isCanceled()) {
						if (ExceptionUtils.stripCompletionException(failure) instanceof TimeoutException) {
							// we simply have not received a response in time. maybe the timeout was
							// very low (initial fast registration attempts), maybe the target endpoint is
							// currently down.
							if (log.isDebugEnabled()) {
								log.debug("Registration at {} ({}) attempt {} timed out after {} ms",
									targetName, targetAddress, attempt, timeoutMillis);
							}

							long newTimeoutMillis = Math.min(2 * timeoutMillis, retryingRegistrationConfiguration.getMaxRegistrationTimeoutMillis());
							register(gateway, attempt + 1, newTimeoutMillis);
						}
						else {
							// a serious failure occurred. we still should not give up, but keep trying
							log.error("Registration at {} failed due to an error", targetName, failure);
							log.info("Pausing and re-attempting registration in {} ms", retryingRegistrationConfiguration.getErrorDelayMillis());

							registerLater(gateway, 1, retryingRegistrationConfiguration.getInitialRegistrationTimeoutMillis(), retryingRegistrationConfiguration.getErrorDelayMillis());
						}
					}
				},
				rpcService.getExecutor());
		}
		catch (Throwable t) {
			completionFuture.completeExceptionally(t);
			cancel();
		}
	}
524118
@SafeVarargs
    public final ServerListenerBuilder addStoppedCallbacks(Consumer<? super Server>... consumers) {
        return addStoppedCallbacks(Arrays.asList(consumers));
    }
1288570
@Override
    public boolean registerConsumerSetMonitor(
                                              SIDestinationAddress destinationAddress,
                                              String discriminatorExpression,
                                              ConsumerSetChangeCallback callback)
                    throws SIResourceException,
                    SINotPossibleInCurrentConfigurationException,
                    SIConnectionUnavailableException,
                    SIConnectionDroppedException,
                    SIIncorrectCallException,
                    SICommandInvocationFailedException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "registerConsumerSetMonitor",
                        new Object[] { destinationAddress,
                                      discriminatorExpression,
                                      callback });

        // Check that a listener has been specified
        if (callback == null)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_CONSUMERSETCHANGECALLBACK_CWSIP0667", null, null));
        } else if (destinationAddress == null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_DESTINATIONADDRESS_CWSIP0668", null, null));

        } else if (discriminatorExpression == null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_DISCRIMINATOREXPRESSION_CWSIP0669", null, null));

        }
        boolean areConsumers = false;
        try
        {
            DestinationHandler topicSpace =
                            _destinationManager.getDestination((JsDestinationAddress) destinationAddress, false);

            areConsumers =
                            _messageProcessor.
                                            getMessageProcessorMatching().
                                            registerConsumerSetMonitor(topicSpace,
                                                                       discriminatorExpression,
                                                                       this,
                                                                       callback);
        } catch (Exception e)
        {
            // FFDC
            FFDCFilter
                            .processException(
                                              e,
                                              "com.ibm.ws.sib.processor.impl.ConnectionImpl.registerConsumerSetMonitor",
                                              "1:7378:1.347.1.25",
                                              this);

            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.exception(tc, e);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIErrorException");
            //SIErrorException is RuntimeExcetion hence catching this exception in StaticCATConnection
            throw new SIErrorException(e);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "registerConsumerSetMonitor", Boolean.valueOf(areConsumers));
        return areConsumers;
    }
1090409
@Override
    public RuntimeEventListener register(Controllable controllable, ControllableType type)
                    throws AlreadyRegisteredException, SIBExceptionInvalidValue {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "register", new Object[] { controllable, type });

        RuntimeEventListener retValue = null;
        if (type == ControllableType.QUEUE_POINT) {
            createQueueMBean(controllable);

        }
        else if (type == ControllableType.PUBLICATION_POINT) {
            createTopicMBean(controllable);
        }
        else if (type == ControllableType.SUBSCRIPTION_POINT) {
            createSubscriberMBean(controllable);
        }
        else
        {
            String reason = "Invalid ControllableType of " + type.toString();
            final SIBExceptionInvalidValue e =
                            new SIBExceptionInvalidValue(
                                            nls.getFormattedMessage("INTERNAL_ERROR_SIAS0003", new Object[] { reason }, reason));
            FFDCFilter.processException(e, CLASS_NAME + ".register", "PROBE_ID_10", this);
            SibTr.exception(tc, e);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "register", e);
            throw e;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "register", new Object[] { retValue });
        return retValue;
    }
291535
void registerCallback(AsynchConsumerCallback callback)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "registerCallback", callback);

    asynchConsumerCallback = callback;

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "registerCallback");
  }
513676
public static void registerConverters(Set<?> converters, ConverterRegistry registry) {
		if (converters != null) {
			for (Object converter : converters) {
				if (converter instanceof GenericConverter) {
					registry.addConverter((GenericConverter) converter);
				}
				else if (converter instanceof Converter<?, ?>) {
					registry.addConverter((Converter<?, ?>) converter);
				}
				else if (converter instanceof ConverterFactory<?, ?>) {
					registry.addConverterFactory((ConverterFactory<?, ?>) converter);
				}
				else {
					throw new IllegalArgumentException("Each converter object must implement one of the " +
							"Converter, ConverterFactory, or GenericConverter interfaces");
				}
			}
		}
	}
327638
public void stoppableConsumerSessionStopped () {
     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "stoppableConsumerSessionStopped");

     try {
       stopInternal(false); // No need to notify our peer as the stop came from our peer
     } catch (Exception e) {
       FFDCFilter.processException(e, CLASS_NAME + ".processAsyncSessionStoppedCallback", CommsConstants.CONSUMERSESSIONPROXY_SESSION_STOPPED_03, this);
     }

     final AsynchConsumerCallback asynchConsumerCallback = proxyQueue.getAsynchConsumerCallback();
     if (asynchConsumerCallback instanceof StoppableAsynchConsumerCallback) {
       final StoppableAsynchConsumerCallback stoppableAsynchConsumerCallback = (StoppableAsynchConsumerCallback)asynchConsumerCallback;
       // Protect ourselves from the application as far as possible then call the application
       try {
         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "** Calling application StoppableAsynchConsumerCallback");
         stoppableAsynchConsumerCallback.consumerSessionStopped();
       } catch (Throwable e) {
         FFDCFilter.processException(e, CLASS_NAME + ".stoppableConsumerSessionStopped", CommsConstants.CONSUMERSESSIONPROXY_SESSION_STOPPED_01, this);
       }
     } else {
       if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "asynchConsumerCallback not an instance of StoppableAsynchConsumerCallback is an instance of "+ asynchConsumerCallback.getClass().getName());
       SIErrorException e = new SIErrorException(nls.getFormattedMessage("WRONG_CLASS_CWSICO8022", new Object[] {asynchConsumerCallback.getClass().getName()}, null));
       FFDCFilter.processException(e, CLASS_NAME + ".processAsyncSessionStoppedCallback", CommsConstants.CONSUMERSESSIONPROXY_SESSION_STOPPED_02, this);
       SibTr.error(tc, "WRONG_CLASS_CWSICO8022", e);
       throw e;
     }

     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "stoppableConsumerSessionStopped");
   }
1329714
public void iterateAllRegisteredTypeConverters (@Nonnull final ITypeConverterCallback aCallback)
  {
    // Create a copy of the map
    final Map <Class <?>, Map <Class <?>, ITypeConverter <?, ?>>> aCopy = m_aRWLock.readLocked ( () -> new CommonsHashMap <> (m_aConverter));

    // And iterate the copy
    outer: for (final Map.Entry <Class <?>, Map <Class <?>, ITypeConverter <?, ?>>> aSrcEntry : aCopy.entrySet ())
    {
      final Class <?> aSrcClass = aSrcEntry.getKey ();
      for (final Map.Entry <Class <?>, ITypeConverter <?, ?>> aDstEntry : aSrcEntry.getValue ().entrySet ())
        if (aCallback.call (aSrcClass, aDstEntry.getKey (), aDstEntry.getValue ()).isBreak ())
          break outer;
    }
  }
1029476
private void addConsumerToRegisteredMonitorMap(
    MonitoredConsumer mc,
    ArrayList monitorList,
    Map registeredMonitors)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.entry(
          tc,
          "addConsumerToRegisteredMonitorMap",
          new Object[] { registeredMonitors,
                         mc,
                         monitorList });
    // Finally, add this consumer to the appropriate places in the registered
    // consumer monitors tables
    Iterator i = monitorList.iterator();
    while (i.hasNext())
    {
      String tExpr = (String)i.next();

      // Get the next monitor
      RegisteredCallbacks rMonitor = (RegisteredCallbacks)registeredMonitors.get(tExpr);

      // Retrieve the current list of consumers
      ArrayList matchingConsumerList = rMonitor.getMatchingConsumers();

      // There is no need to drive any callbacks as we are placing subscriptions
      // in the registrations table that correspond to existing subscriptions.
      // So simply add the new subscription to the list.
      matchingConsumerList.add(mc);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "addConsumerToRegisteredMonitorMap");

  }
682924
public void addConsumerToRegisteredMonitors(
    MonitoredConsumer mc,
    ArrayList exactMonitorList,
    ArrayList wildcardMonitorList)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.entry(
          tc,
          "addConsumerToRegisteredMonitors",
          new Object[] { mc,
                         exactMonitorList,
                         wildcardMonitorList });
    // Add consumer to correct places in the maps
    addConsumerToRegisteredMonitorMap(mc, exactMonitorList, _registeredExactConsumerMonitors);

    // Now process the wildcard monitor list
    addConsumerToRegisteredMonitorMap(mc, wildcardMonitorList, _registeredWildcardConsumerMonitors);

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "addConsumerToRegisteredMonitors");

  }
500254
public synchronized MessageRouterConfigListenerImpl register(BundleContext bundleContext) {
        if (serviceRegistration != null) {
            return this; // Already registered.
        }

        Dictionary<String, Object> props = new Hashtable<String, Object>(1);
        props.put(Constants.SERVICE_VENDOR, "IBM");

        // register the ManagedService that will handle the config
        serviceRegistration = bundleContext.registerService(MessageRouterConfigListener.class, this, props);
        return this;
    }