<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.messagetransport.passthrough;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import com.nokia.dempsy.Adaptor;
import com.nokia.dempsy.executor.DempsyExecutor;
import com.nokia.dempsy.internal.util.SafeString;
import com.nokia.dempsy.messagetransport.Destination;
import com.nokia.dempsy.messagetransport.Listener;
import com.nokia.dempsy.messagetransport.MessageTransportException;
import com.nokia.dempsy.messagetransport.OverflowHandler;
import com.nokia.dempsy.messagetransport.Receiver;
import com.nokia.dempsy.messagetransport.Sender;
import com.nokia.dempsy.messagetransport.SenderFactory;
import com.nokia.dempsy.messagetransport.Transport;
import com.nokia.dempsy.monitoring.StatsCollector;

/**
 * This transport is basically an inVM transport where the call-stack becomes 
 * the transport mechanism. It's primarily for testing when completely deterministic 
 * behavior is required.
 * 
 * When the 'send' call simply directly calls the {@link Listener} on the receive
 * side. As a result the thread will basically block in the 'send' until that message 
 * is handled, by the {@link Listener} in the same thread the 'send' was called in.
 * 
 * When an inVM Dempsy application is using this transport the effect is that the
 * initial {@link Adaptor}'s call to 'send' will not return until that message is
 * processed and any resulting messages are passed on to the next Mp in the processing 
 * chain and those are processed.
 */
public class PassthroughTransport implements Transport
{
   private boolean failFast = true;
   
   @Override
   public SenderFactory createOutbound(DempsyExecutor executor, StatsCollector statsCollector)
   {
      return new SenderFactory()
      {
         private volatile boolean isStopped = false;

         @Override
         public Sender getSender(Destination destination) throws MessageTransportException
         {
            if (isStopped == true)
               throw new MessageTransportException("getSender called for the destination " + SafeString.valueOf(destination) + 
                     " on a stopped " + SafeString.valueOfClass(this));

            return ((PassthroughDestination)destination).sender;
         }
         
         @Override
         public void stop() { isStopped = true; }
      };
   }
   
   @Override
   public Receiver createInbound(DempsyExecutor executor) throws MessageTransportException
   {
      return new Receiver()
      {
         List<Listener> listeners = new CopyOnWriteArrayList<Listener>();
         
         Sender sender = new Sender()
         {
            @Override
            public void send(byte[] messageBytes) throws MessageTransportException
            {
               for (Listener listener : listeners)
                  listener.onMessage(messageBytes, failFast);
            }
         };
         
         PassthroughDestination destination = new PassthroughDestination(sender);
         
         @Override
         public void setListener(Listener listener) throws MessageTransportException
         {
            listeners.add(listener);
         }
         
         @Override
         public Destination getDestination() throws MessageTransportException
         {
            return destination;
         }
         
         @Override
         public void stop() {}
         
         @Override
         public void start() {}
         
         @Override
         public void setStatsCollector(StatsCollector statsCollector) {}
         
      };<commit_msg>The Passthrough transport now handles the StatsCollector correctly.<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.messagetransport.passthrough;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import com.nokia.dempsy.Adaptor;
import com.nokia.dempsy.executor.DempsyExecutor;
import com.nokia.dempsy.internal.util.SafeString;
import com.nokia.dempsy.messagetransport.Destination;
import com.nokia.dempsy.messagetransport.Listener;
import com.nokia.dempsy.messagetransport.MessageTransportException;
import com.nokia.dempsy.messagetransport.OverflowHandler;
import com.nokia.dempsy.messagetransport.Receiver;
import com.nokia.dempsy.messagetransport.Sender;
import com.nokia.dempsy.messagetransport.SenderFactory;
import com.nokia.dempsy.messagetransport.Transport;
import com.nokia.dempsy.monitoring.StatsCollector;

/**
 * This transport is basically an inVM transport where the call-stack becomes 
 * the transport mechanism. It's primarily for testing when completely deterministic 
 * behavior is required.
 * 
 * When the 'send' call simply directly calls the {@link Listener} on the receive
 * side. As a result the thread will basically block in the 'send' until that message 
 * is handled, by the {@link Listener} in the same thread the 'send' was called in.
 * 
 * When an inVM Dempsy application is using this transport the effect is that the
 * initial {@link Adaptor}'s call to 'send' will not return until that message is
 * processed and any resulting messages are passed on to the next Mp in the processing 
 * chain and those are processed.
 */
public class PassthroughTransport implements Transport
{
   private boolean failFast = true;
   
   @Override
   public SenderFactory createOutbound(DempsyExecutor executor, final StatsCollector statsCollector)
   {
      return new SenderFactory()
      {
         private volatile boolean isStopped = false;
         private StatsCollector sc = statsCollector;

         @Override
         public Sender getSender(Destination destination) throws MessageTransportException
         {
            if (isStopped == true)
               throw new MessageTransportException("getSender called for the destination " + SafeString.valueOf(destination) + 
                     " on a stopped " + SafeString.valueOfClass(this));

            PassthroughSender ret = (PassthroughSender)(((PassthroughDestination)destination).sender);
            ret.statsCollector = sc;
            return ret;
         }
         
         @Override
         public void stop() { isStopped = true; }
      };
   }
   
   private class PassthroughSender implements Sender
   {
      StatsCollector statsCollector = null;
      List<Listener> listeners;
      
      private PassthroughSender(List<Listener> listeners) { this.listeners = listeners; }

      @Override
      public void send(byte[] messageBytes) throws MessageTransportException
      {
         for (Listener listener : listeners)
         {
            if (statsCollector != null) statsCollector.messageSent(messageBytes);
            listener.onMessage(messageBytes, failFast);
         }
      }
   }
   
   @Override
   public Receiver createInbound(DempsyExecutor executor) throws MessageTransportException
   {
      return new Receiver()
      {
         List<Listener> listeners = new CopyOnWriteArrayList<Listener>();
         
         PassthroughDestination destination = new PassthroughDestination(new PassthroughSender(listeners));
         
         @Override
         public void setListener(Listener listener) throws MessageTransportException
         {
            listeners.add(listener);
         }
         
         @Override
         public Destination getDestination() throws MessageTransportException
         {
            return destination;
         }
         
         @Override
         public void stop() {}
         
         @Override
         public void start() {}
         
         @Override
         public void setStatsCollector(StatsCollector statsCollector) { }
         
      };
{'fraction_non_alphanumeric': 0.06857530836256648, 'fraction_numerical': 0.0018105691976915243, 'mean_word_length': 2.884835164835165, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '4338097', 'n_tokens_mistral': 2286, 'n_tokens_neox': 2208, 'n_words': 729}