<commit_before>/*
 * Copyright 2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.nokia.dempsy;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.inject.Inject;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MarkerFactory;

import com.nokia.dempsy.cluster.ClusterInfoException;
import com.nokia.dempsy.cluster.ClusterInfoSession;
import com.nokia.dempsy.cluster.ClusterInfoSessionFactory;
import com.nokia.dempsy.config.ApplicationDefinition;
import com.nokia.dempsy.config.ClusterDefinition;
import com.nokia.dempsy.config.ClusterId;
import com.nokia.dempsy.container.MpContainer;
import com.nokia.dempsy.internal.util.SafeString;
import com.nokia.dempsy.messagetransport.Destination;
import com.nokia.dempsy.messagetransport.Receiver;
import com.nokia.dempsy.messagetransport.Transport;
import com.nokia.dempsy.monitoring.StatsCollector;
import com.nokia.dempsy.monitoring.StatsCollectorFactory;
import com.nokia.dempsy.output.OutputExecuter;
import com.nokia.dempsy.router.CurrentClusterCheck;
import com.nokia.dempsy.router.Router;
import com.nokia.dempsy.router.RoutingStrategy;
import com.nokia.dempsy.serialization.Serializer;

/**
 * This class is the master orchestrator of the setup of a VM running the application.
 * It is intended to have the ApplicationDefinitions autowired in so it picks up
 * each one.
 */
public class Dempsy
{
   static Logger logger = LoggerFactory.getLogger(Dempsy.class);
   
   /**
    * There is an instance of an application for every ApplicationDefinition
    * autowired into the Dempsy instance. In most real cases this will be only
    * one.
    */
   public class Application
   {
      /**
       * Currently a "Dempsy.Application.Cluster" has no utility since it only has a single Node.
       * This may change
       */
      public class Cluster
      {
         /**
          * A Node is essentially an {@link MpContainer} with all of the attending infrastructure.
          * Currently a Node is instantiated within the Dempsy orchestrator as a one to one with the
          * {@link Cluster}.
          */
         public class Node
         {
            protected ClusterDefinition clusterDefinition;
            
            Router router = null;
            private MpContainer container = null;
            RoutingStrategy.Inbound strategyInbound = null;
            List<Class<?>> acceptedMessageClasses = null;
            Receiver receiver = null;
            StatsCollector statsCollector = null;
            
            private Node(ClusterDefinition clusterDefinition) { this.clusterDefinition = clusterDefinition; }
            
            @SuppressWarnings("unchecked")
            private void start() throws DempsyException
            {
               try
               {
                  ClusterInfoSession clusterSession = clusterSessionFactory.createSession();
                  ClusterId currentClusterId = clusterDefinition.getClusterId();
                  router = new Router(clusterDefinition.getParentApplicationDefinition());
                  router.setCurrentCluster(currentClusterId);
                  router.setClusterSession(clusterSession);
                  router.setDefaultSenderFactory(transport.createOutbound());
                  
                  container = new MpContainer(currentClusterId);
                  container.setDispatcher(router);
                  Object messageProcessorPrototype = clusterDefinition.getMessageProcessorPrototype();
                  if (messageProcessorPrototype != null)
                  {
                    preInitializePrototype(messageProcessorPrototype);
                     container.setPrototype(messageProcessorPrototype);
                  }
                  acceptedMessageClasses = getAcceptedMessages(clusterDefinition);
                  
                  Serializer<Object> serializer = (Serializer<Object>)clusterDefinition.getSerializer();
                  if (serializer != null)
                     container.setSerializer(serializer);
                  
                  // there is only a reciever if we have an Mp (that is, we aren't an adaptor) and start accepting messages 
                  Destination thisDestination = null;
                  if (messageProcessorPrototype != null && acceptedMessageClasses != null && acceptedMessageClasses.size() > 0)
                  {
                     receiver = transport.createInbound();
                     receiver.setListener(container);
                     thisDestination = receiver.getDestination();
                  }

                  StatsCollectorFactory statsFactory = (StatsCollectorFactory)clusterDefinition.getStatsCollectorFactory();
                  if (statsFactory != null)
                  {
                     statsCollector = statsFactory.createStatsCollector(currentClusterId, 
                           receiver != null ? thisDestination : null);
                     router.setStatsCollector(statsCollector);
                     container.setStatCollector(statsCollector);
                  }
                  
                  RoutingStrategy strategy = (RoutingStrategy)clusterDefinition.getRoutingStrategy();
                  
                  KeySource<?> keySource = clusterDefinition.getKeySource();
                  if (keySource != null)
                     container.setKeySource(keySource);

                  // there is only an inbound strategy if we have an Mp (that is, we aren't an adaptor) and
                  // we actually accept messages
                  if (messageProcessorPrototype != null && acceptedMessageClasses != null && acceptedMessageClasses.size() > 0)
                     strategyInbound = strategy.createInbound(clusterSession,currentClusterId,acceptedMessageClasses, thisDestination,container);
                  
                  // this can fail because of down cluster manager server ... but it should eventually recover.
                  try { router.initialize(); }
                  catch (ClusterInfoException e)
                  {
                     logger.warn("Strategy failed to initialize. Continuing anyway. The cluster manager issue will be resolved automatically.",e);
                  }
                  
                  Adaptor adaptor = clusterDefinition.isRouteAdaptorType() ? clusterDefinition.getAdaptor() : null;<commit_msg>[fix] The StatsCollector was never set on the receiver so shed messages due to busy transports where not being recorded.<commit_after>/*
 * Copyright 2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.nokia.dempsy;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.inject.Inject;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MarkerFactory;

import com.nokia.dempsy.cluster.ClusterInfoException;
import com.nokia.dempsy.cluster.ClusterInfoSession;
import com.nokia.dempsy.cluster.ClusterInfoSessionFactory;
import com.nokia.dempsy.config.ApplicationDefinition;
import com.nokia.dempsy.config.ClusterDefinition;
import com.nokia.dempsy.config.ClusterId;
import com.nokia.dempsy.container.MpContainer;
import com.nokia.dempsy.internal.util.SafeString;
import com.nokia.dempsy.messagetransport.Destination;
import com.nokia.dempsy.messagetransport.Receiver;
import com.nokia.dempsy.messagetransport.Transport;
import com.nokia.dempsy.monitoring.StatsCollector;
import com.nokia.dempsy.monitoring.StatsCollectorFactory;
import com.nokia.dempsy.output.OutputExecuter;
import com.nokia.dempsy.router.CurrentClusterCheck;
import com.nokia.dempsy.router.Router;
import com.nokia.dempsy.router.RoutingStrategy;
import com.nokia.dempsy.serialization.Serializer;

/**
 * This class is the master orchestrator of the setup of a VM running the application.
 * It is intended to have the ApplicationDefinitions autowired in so it picks up
 * each one.
 */
public class Dempsy
{
   static Logger logger = LoggerFactory.getLogger(Dempsy.class);
   
   /**
    * There is an instance of an application for every ApplicationDefinition
    * autowired into the Dempsy instance. In most real cases this will be only
    * one.
    */
   public class Application
   {
      /**
       * Currently a "Dempsy.Application.Cluster" has no utility since it only has a single Node.
       * This may change
       */
      public class Cluster
      {
         /**
          * A Node is essentially an {@link MpContainer} with all of the attending infrastructure.
          * Currently a Node is instantiated within the Dempsy orchestrator as a one to one with the
          * {@link Cluster}.
          */
         public class Node
         {
            protected ClusterDefinition clusterDefinition;
            
            Router router = null;
            private MpContainer container = null;
            RoutingStrategy.Inbound strategyInbound = null;
            List<Class<?>> acceptedMessageClasses = null;
            Receiver receiver = null;
            StatsCollector statsCollector = null;
            
            private Node(ClusterDefinition clusterDefinition) { this.clusterDefinition = clusterDefinition; }
            
            @SuppressWarnings("unchecked")
            private void start() throws DempsyException
            {
               try
               {
                  ClusterInfoSession clusterSession = clusterSessionFactory.createSession();
                  ClusterId currentClusterId = clusterDefinition.getClusterId();
                  router = new Router(clusterDefinition.getParentApplicationDefinition());
                  router.setCurrentCluster(currentClusterId);
                  router.setClusterSession(clusterSession);
                  router.setDefaultSenderFactory(transport.createOutbound());
                  
                  container = new MpContainer(currentClusterId);
                  container.setDispatcher(router);
                  Object messageProcessorPrototype = clusterDefinition.getMessageProcessorPrototype();
                  if (messageProcessorPrototype != null)
                  {
                    preInitializePrototype(messageProcessorPrototype);
                     container.setPrototype(messageProcessorPrototype);
                  }
                  acceptedMessageClasses = getAcceptedMessages(clusterDefinition);
                  
                  Serializer<Object> serializer = (Serializer<Object>)clusterDefinition.getSerializer();
                  if (serializer != null)
                     container.setSerializer(serializer);
                  
                  // there is only a reciever if we have an Mp (that is, we aren't an adaptor) and start accepting messages 
                  Destination thisDestination = null;
                  if (messageProcessorPrototype != null && acceptedMessageClasses != null && acceptedMessageClasses.size() > 0)
                  {
                     receiver = transport.createInbound();
                     receiver.setListener(container);
                     thisDestination = receiver.getDestination();
                  }

                  StatsCollectorFactory statsFactory = (StatsCollectorFactory)clusterDefinition.getStatsCollectorFactory();
                  if (statsFactory != null)
                  {
                     statsCollector = statsFactory.createStatsCollector(currentClusterId, 
                           receiver != null ? thisDestination : null);
                     router.setStatsCollector(statsCollector);
                     container.setStatCollector(statsCollector);
                     if (receiver != null)
                        receiver.setStatsCollector(statsCollector);
                  }
                  
                  RoutingStrategy strategy = (RoutingStrategy)clusterDefinition.getRoutingStrategy();
                  
                  KeySource<?> keySource = clusterDefinition.getKeySource();
                  if (keySource != null)
                     container.setKeySource(keySource);

                  // there is only an inbound strategy if we have an Mp (that is, we aren't an adaptor) and
                  // we actually accept messages
                  if (messageProcessorPrototype != null && acceptedMessageClasses != null && acceptedMessageClasses.size() > 0)
                     strategyInbound = strategy.createInbound(clusterSession,currentClusterId,acceptedMessageClasses, thisDestination,container);
                  
                  // this can fail because of down cluster manager server ... but it should eventually recover.
                  try { router.initialize(); }
                  catch (ClusterInfoException e)
                  {
                     logger.warn("Strategy failed to initialize. Continuing anyway. The cluster manager issue will be resolved automatically.",e);
                  }
                  
                  Adaptor adaptor = clusterDefinition.isRouteAdaptorType() ? clusterDefinition.getAdaptor() : null;
{'fraction_non_alphanumeric': 0.07076858462830743, 'fraction_numerical': 0.0018199636007279855, 'mean_word_length': 2.2559252506836827, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4667004', 'n_tokens_mistral': 3382, 'n_tokens_neox': 3229, 'n_words': 932}