public ThreadPoolConsumerPool(IServiceBus bus, IObjectBuilder objectBuilder, Pipe eventAggregator, TimeSpan receiveTimeout)
		{
			_objectBuilder = objectBuilder;
			_receiveTimeout = receiveTimeout;
			_eventAggregator = eventAggregator;
			_bus = bus;
		}

        // TODO: Eliminate these AsPipe/AsConitnuable switches... framework-wide
        private void mapCore(string urlPrefix, Func<Pipe, Pipe, Pipe> mapFunc, Pipe cont)
        {
            var err = HttpErrors.MethodNotAllowed();
              var notMethod = _mappings.FindMapping(urlPrefix, ifNotFound: err);

              _mappings[urlPrefix] = mapFunc(cont, notMethod);
        }

        public void Merge(
            FileInfo           source,
            Pipe<CilDocumentSyntax> compile,
            FileInfo           destination)
        {
            CilSyntax context = CilBackend.CreateCompiler(null);
            context
            .BeginDocument()
                .AssemblyRewrite(source.FullName)
                    .CustomAttribute(
                        context.Types.Import(
                            typeof(DerivedAssemblyMarker)))
                .EndAssembly()
                .Do(compile)
            .EndDocument();

            ((IAssemblyResolverParameters)context).AddSearchDirectory(destination.DirectoryName);

            var writer = context as IAssemblyWriter;
            if (writer == null)
            {
                throw new InvalidOperationException("Backend does not support assembly writing");
            }

            writer.Write(destination.FullName);
        }

    public override void GenerateItems(Pipe pipe)
    {
        float start = pipe.pipeSegmentCount + 0.5f;
        //float direction = Random.value < 0.5f ? 1f : -1f;

        //float angleStep = pipe.CurveAngle / pipe.CurveSegmentCount;

        float direction = 0.30f;

        float angleStep = 2.4f;
        for (int i = 0; i < 1; i++)
        {
            if(i>1)
            {

                PipeItem item = Instantiate<PipeItem>(
                itemPrefabs[Random.Range(0, itemPrefabs.Length)].GetComponent<PipeItem>());
                float pipeRotation =
                    (start + i * direction) * 360f / pipe.pipeSegmentCount;
                item.Position(pipe, angleStep, pipeRotation);

            }
            else
            {

                PipeItem item = Instantiate<PipeItem>(
                itemPrefabs[Random.Range(0, itemPrefabs.Length)].GetComponent<PipeItem>());
                float pipeRotation =
                    ((start * 7.5f) + i * direction) * 360f / pipe.pipeSegmentCount;
                item.Position(pipe, angleStep, pipeRotation);

            }

        }
    }

        public static void PumpingRound()
        {
            Pump<TextMessage> pump = new Pump<TextMessage>();
            pump.Initialize();
            pump.Interval = 1;
            
            Pipe<TextMessage> p1 = new Pipe<TextMessage>();
            Pipe<TextMessage> p2 = new Pipe<TextMessage>();
            Pipe<TextMessage> p3 = new Pipe<TextMessage>();
            Pipe<TextMessage> p4 = new Pipe<TextMessage>();

            p1.Initialize();
            p2.Initialize();
            p3.Initialize();
            p4.Initialize();

            pump.AddFlow(p1, p2);
            pump.AddFlow(p2, p3);
            pump.AddFlow(p3, p4);
            pump.AddFlow(p4, p1);

            p1.AddInputNotify(new Notify(NotifyOutput));
            p2.AddInputNotify(new Notify(NotifyOutput2));
            p3.AddInputNotify(new Notify(NotifyOutput));
            p4.AddInputNotify(new Notify(NotifyOutput2));
            
            p1.Push(new TextMessage("bla bla"));

            pump.Start();            

            Console.ReadLine();
        }

 public BasicApiProvider_1_4(IMapper mapper, Pipe eventAggregator)
 {
     Helper.GuardNotNull(mapper);
     Helper.GuardNotNull(eventAggregator);
     _mapper = mapper;
     _eventAggregator = eventAggregator;
 }

        public Graph CreateGraph(Pipe pipe)
        {
            var visitor = new GraphPipelineVisitor();
            visitor.Visit(pipe);

            return CreateGraph(visitor.Vertices, visitor.Edges);
        }

 public void Position(Pipe pipe, float curveRotation, float ringRotation)
 {
     transform.SetParent(pipe.transform, false);
     transform.localRotation = Quaternion.Euler(0f, 0f, -curveRotation);
     rotater.localPosition = new Vector3(0f, pipe.CurveRadius);
     rotater.localRotation = Quaternion.Euler(ringRotation, 0f, 0f);
 }

        protected virtual Pipe Visit(Pipe pipe)
        {
            if (pipe == null)
                return pipe;

            switch (pipe.SegmentType)
            {
                case PipeSegmentType.End:
                    return VisitEnd((EndSegment)pipe);

                case PipeSegmentType.Filter:
                    return VisitFilter((FilterSegment)pipe);

                case PipeSegmentType.Input:
                    return VisitInput((InputSegment)pipe);

                case PipeSegmentType.MessageConsumer:
                    return VisitMessageConsumer((MessageConsumerSegment)pipe);

                case PipeSegmentType.RecipientList:
                    return VisitRecipientList((RecipientListSegment)pipe);

                case PipeSegmentType.Interceptor:
                    return VisitInterceptor((InterceptorSegment)pipe);

                default:
                    throw new ArgumentException("The pipe node is not a known type: " + pipe.SegmentType,
                                                "pipeline");
            }
        }

        private Pipe displayIndex(Pipe notFound)
        {
            var specialFolders = new[] { ".", ".." };

              return (ctx, next) =>
              {
            // validate path
            var curPath = Path.Combine(_basePath, ctx.Request.Path.Substring(1));
            var fullPath = Path.GetFullPath(curPath);

            if (!Directory.Exists(curPath) || !fullPath.StartsWith(_basePath)) {
              notFound(ctx, next);
              return;
            }

            // generate list of files and folders to display in the index
            var entries = Directory
              .GetFiles(curPath)
              .Concat(Directory.GetDirectories(curPath))
              .Select(path => path.Substring(_basePath.Length));

            if (curPath != _basePath)
              entries = entries.Concat(specialFolders);

            // render the index page
            var html = _template.RenderIndex(curPath, entries.ToArray());
            Static.String(Mime.Text.Html, html)(ctx, next);
              };
        }

 void spawnPipes()
 {
     lastPipes = (Pipe)Instantiate(pipes);
     lastPipes.p = p;
     lastPipes.speed = pipeSpeed;
     lastPipes.transform.position = new Vector3(12,Random.Range(minHeight,maxHeight),0);
 }

 public GdbProtocol(Pipe pipe)
 {
     mPipe = pipe;
     mPipe.PipeReceiveEvent += RecvFired;
     mTimer.Elapsed += ResendTimer;
     mTimer.Start();
 }

		internal BlockingPipe(Pipe pipe)
			: base(pipe.Loop)
		{
			Handle = pipe;
			Stream = pipe;
			Pipe = pipe;
		}

        public static IServer Connect(this IContainer container,
            string host, int port, Pipe pipes, params IService[] services)
        {
            var broker = container.BuildServicesBroker(services);
              var handler = container.BuildRequestHandler(pipes, broker);

              return container.BuildServer(host, port, handler);
        }

        public void TimeoutTest()
        {
            var pipe = new Pipe {OutputStream = {ReadTimeout = 0}};
            Assert.Throws<TimeoutException>(() => pipe.OutputStream.ReadByte());

            pipe.WriteText(new string('a', 2048));
            Assert.AreEqual(new string('a', 2048), pipe.ReadTextAsync(2048).Result);
        }

 // Use this for initialization
 void Start()
 {
     world = pipeSystem.transform.parent;
     rotater = transform.GetChild(0);
     currentPipe = pipeSystem.SetupFirstPipe();
     deltaToRotation = 360f / (2f * Mathf.PI * currentPipe.CurveRadius);
     SetupCurrentPipe();
 }

        public void Setup()
        {
            Input = PipeSegment.Input(PipeSegment.End<object>());
            BeforeCalled = new ManualResetEvent(false);
            AfterCalled = new ManualResetEvent(false);

            EstablishContext();
        }

		public StraightThroughPipelineRunner()
		{
			Pipe consumer = PipeSegment.Consumer<ClaimModified>(m => Interlocked.Increment(ref _count));

			_input = PipeSegment.Input(consumer);

			_message = new ClaimModified();
		}

 public void Disconnect(Pipe pipe)
 {
     if(Pipe != pipe)
     {
         Debug.LogWarning("Tried to disconnect unconnected pipe");
         return;
     }
     Pipe = null;
 }

 private Pipe indexCheck(Pipe onFolder, Pipe onFile)
 {
     return (ctx, next) =>
       {
     var path = mapPath(ctx.Request.Path);
     (string.IsNullOrEmpty(path) || Directory.Exists(path) ? onFolder : onFile)
       (ctx, next);
       };
 }