blob: 2593b44b54284236100e5c3f4753258990c72a16 [file] [log] [blame]
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
using System;
using System.Collections.Generic;
using Org.Apache.REEF.Utilities.Logging;
using Org.Apache.REEF.Wake.Util;
namespace Org.Apache.REEF.Wake.Examples.P2p
/// <summary>Performs a Pull-to-Push conversion in Wake.</summary>
/// <remarks>
/// Performs a Pull-to-Push conversion in Wake.
/// The class pulls from a set of event sources, and pushes to a single
/// EventHandler. If the downstream event handler blocks, this will block,
/// providing a simple rate limiting scheme.
/// The EventSources are managed in a basic Queue.
/// </remarks>
public sealed class Pull2Push<T> : IStartable, IDisposable
private static readonly Logger LOGGER = Logger.GetLogger(typeof(Pull2Push<T>));
private readonly IEventHandler<T> _output;
private readonly Queue<IEventSource<T>> _sources = new Queue<IEventSource<T>>();
private bool _closed = false;
/// <summary>
/// Constructs a new Pull2Push object
/// </summary>
/// <param name="output">
/// the EventHandler that receives the messages from this
/// Pull2Push.
/// </param>
public Pull2Push(IEventHandler<T> output)
// The downstream EventHandler
// The upstream event sources
_output = output;
/// <summary>Registers an event source.</summary>
/// <param name="source">
/// The source that will be added to the queue of this
/// Pull2Push
/// </param>
public void Register(IEventSource<T> source)
/// <summary>Executes the message loop.</summary>
public void Start()
while (!_closed)
// Grab the next available message source, if any
IEventSource<T> nextSource = _sources.Dequeue();
if (null != nextSource)
// Grab the next message from that source, if any
T message = nextSource.GetNext();
if (null != message)
// Add the source to the end of the queue again.
// Send the message. Note that this may block depending on the underlying EventHandler.
// The message source has returned null as the next message. We drop the message source in that case.
LOGGER.Log(Level.Info, "Dropping message source {0} from the queue " + nextSource.ToString());
// No source where available. We could put a wait() here.
public void Dispose()
_closed = true;