Index: XIOS/trunk/src/filter/data_packet.hpp
===================================================================
--- XIOS/trunk/src/filter/data_packet.hpp	(revision 637)
+++ XIOS/trunk/src/filter/data_packet.hpp	(revision 637)
@@ -0,0 +1,47 @@
+#ifndef __XIOS_CDataPacket__
+#define __XIOS_CDataPacket__
+
+#include <boost/shared_ptr.hpp>
+
+#include "array_new.hpp"
+#include "duration.hpp"
+
+namespace xios
+{
+  /*!
+   * A packet corresponds to a timestamped array of data.
+   */
+  struct CDataPacket
+  {
+    /*!
+     * Describes the status associated to a packet
+     */
+    enum StatusCode {
+      NO_ERROR,     //!< No error were encountered when handling the packet
+      END_OF_STREAM //!< Last packet of the stream, does not have data
+    };
+
+    CArray<double, 1> data; //!< Array containing the data
+    Time timestamp;         //!< Timestamp of the data
+    StatusCode status;      //!< Status of the packet
+
+    /*!
+     * Creates a deep copy of the packet.
+     *
+     * \return a deep copy of the packet
+     */
+    CDataPacket* copy() const {
+      CDataPacket* p = new CDataPacket;
+      p->data.resize(data.shape());
+      p->data = data;
+      p->timestamp = timestamp;
+      p->status = status;
+      return p;
+    };
+  }; // struct CDataPacket
+
+  typedef boost::shared_ptr<CDataPacket> CDataPacketPtr;
+  typedef boost::shared_ptr<const CDataPacket> CConstDataPacketPtr;
+} // namespace xios
+
+#endif //__XIOS_CDataPacket__
Index: XIOS/trunk/src/filter/filter.cpp
===================================================================
--- XIOS/trunk/src/filter/filter.cpp	(revision 637)
+++ XIOS/trunk/src/filter/filter.cpp	(revision 637)
@@ -0,0 +1,17 @@
+#include "filter.hpp"
+
+namespace xios
+{
+  CFilter::CFilter(size_t inputSlotsCount, IFilterEngine* engine)
+    : CInputPin(inputSlotsCount)
+    , COutputPin()
+    , engine(engine)
+  { /* Nothing to do */ }
+
+  void CFilter::onInputReady(std::vector<CDataPacketPtr> data)
+  {
+    CDataPacketPtr outputPacket = engine->apply(data);
+    if (outputPacket)
+      deliverOuput(outputPacket);
+  }
+} // namespace xios
Index: XIOS/trunk/src/filter/filter.hpp
===================================================================
--- XIOS/trunk/src/filter/filter.hpp	(revision 637)
+++ XIOS/trunk/src/filter/filter.hpp	(revision 637)
@@ -0,0 +1,41 @@
+#ifndef __XIOS_CFilter__
+#define __XIOS_CFilter__
+
+#include "input_pin.hpp"
+#include "output_pin.hpp"
+#include "filter_engine.hpp"
+
+namespace xios
+{
+  /*!
+   * A generic filter with an input pin and an output pin.
+   * A filter can delegate the internal work to an engine
+   * which may be shared by multiple filter instances.
+   */
+  class CFilter : public CInputPin, public COutputPin
+  {
+    public:
+      /*!
+       * Constructs a filter with the specified number of input slots
+       * and the specified engine.
+       *
+       * \param inputSlotsCount the number of input slots
+       * \param engine the filter engine
+       */
+      CFilter(size_t inputSlotsCount, IFilterEngine* engine);
+
+    protected:
+      IFilterEngine* engine; //!< The filter engine, might be the filter itself
+
+      /*!
+       * Generic implementation of the input pin notification function, processes
+       * the data using the filter engine and passes the resulting packet (if any)
+       * to the downstreams filters.
+       *
+       * \param data a vector of packets corresponding to each slot
+       */
+      void virtual onInputReady(std::vector<CDataPacketPtr> data);
+  }; // class CFilter
+} // namespace xios
+
+#endif //__XIOS_CFilter__
Index: XIOS/trunk/src/filter/filter_engine.hpp
===================================================================
--- XIOS/trunk/src/filter/filter_engine.hpp	(revision 637)
+++ XIOS/trunk/src/filter/filter_engine.hpp	(revision 637)
@@ -0,0 +1,28 @@
+#ifndef __XIOS_IFilterEngine__
+#define __XIOS_IFilterEngine__
+
+#include "data_packet.hpp"
+
+namespace xios
+{
+  /*!
+   * An engine filter processes the data handled by a filter.
+   */
+  class IFilterEngine
+  {
+    public:
+      /*!
+       * Processes the data and returns a new packet if needed.
+       * Temporal filter engines will not return a new packet
+       * for every call being that they accumulate data.
+       * The filter engine is not allowed to modify the data
+       * in place.
+       *
+       * \param data a vector of packets corresponding to each slot
+       * \return a new data packet or null if the engine is accumulating data
+       */
+       CDataPacketPtr virtual apply(std::vector<CDataPacketPtr> data) = 0;
+  }; // class IFilterEngine
+} // namespace xios
+
+#endif //__XIOS_IFilterEngine__
Index: XIOS/trunk/src/filter/input_pin.cpp
===================================================================
--- XIOS/trunk/src/filter/input_pin.cpp	(revision 637)
+++ XIOS/trunk/src/filter/input_pin.cpp	(revision 637)
@@ -0,0 +1,31 @@
+#include "input_pin.hpp"
+#include "exception.hpp"
+
+namespace xios
+{
+  CInputPin::CInputPin(size_t slotsCount)
+    : slotsCount(slotsCount)
+  { /* Nothing to do */ }
+
+  void CInputPin::setInput(size_t inputSlot, CDataPacketPtr packet)
+  {
+    if (inputSlot >= slotsCount)
+      ERROR("void CInputPin::setInput(size_t inputSlot, CDataPacketPtr packet)",
+            "The input slot " << inputSlot << " does not exist.");
+    if (!packet)
+      ERROR("void CInputPin::setInput(size_t inputSlot, CDataPacketPtr packet)",
+            "The packet cannot be null.");
+
+    std::map<Time, InputBuffer>::iterator it = inputs.find(packet->timestamp);
+    if (it == inputs.end())
+      it = inputs.insert(std::make_pair(packet->timestamp, InputBuffer(slotsCount))).first;
+    it->second.slotsFilled++;
+    it->second.packets[inputSlot] = packet;
+
+    if (it->second.slotsFilled == slotsCount)
+    {
+      onInputReady(it->second.packets);
+      inputs.erase(it);
+    }
+  }
+} // namespace xios
Index: XIOS/trunk/src/filter/input_pin.hpp
===================================================================
--- XIOS/trunk/src/filter/input_pin.hpp	(revision 637)
+++ XIOS/trunk/src/filter/input_pin.hpp	(revision 637)
@@ -0,0 +1,71 @@
+#ifndef __XIOS_CInputPin__
+#define __XIOS_CInputPin__
+
+#include <vector>
+#include <map>
+
+#include "data_packet.hpp"
+
+namespace xios
+{
+  /*!
+   * An input pin handles the data packets received by a filter.
+   */
+  class CInputPin
+  {
+    public:
+      /*!
+       * Constructs an input pin with the specified number of slots.
+       *
+       * \param slotsCount the number of slots
+       */
+      CInputPin(size_t slotsCount);
+
+      /*!
+       * Receives a data packet from an upstream filter on
+       * the specified input slot.
+       * The receiving filter takes ownership of the packet.
+       *
+       * \param inputSlot the input slot number
+       * \param packet the data packet to be received
+       */
+      void setInput(size_t inputSlot, CDataPacketPtr packet);
+
+    protected:
+      /*!
+       * Function triggered when all slots have been filled for a specific timestamp.
+       * It should be implemented by the filter class to process the data.
+       *
+       * \param data a vector of packets corresponding to each slot
+       */
+      void virtual onInputReady(std::vector<CDataPacketPtr> data) = 0;
+
+    private:
+      /*!
+       * Helper structure, groups a vector of packets corresponding to each slot
+       * with the number of slots currently filled
+       */
+      struct InputBuffer
+      {
+        size_t slotsFilled; //!< Number of slots currently filled
+        std::vector<CDataPacketPtr> packets; //< Vector of packets corresponding to each slot
+
+        /*!
+         * Initialize an empty input buffer for the specified number of slots.
+         *
+         * \param slotsCount the number of slots
+         */
+        InputBuffer(size_t slotsCount)
+          : slotsFilled(0)
+          , packets(slotsCount)
+        { /* Nothing to do */ };
+      };
+
+      size_t slotsCount; //!< The number of slots
+
+      //! Input buffer, store the packets until all slots are full for a timestep
+      std::map<Time, InputBuffer> inputs;
+  }; // class CInputPin
+} // namespace xios
+
+#endif //__XIOS_CInputPin__
Index: XIOS/trunk/src/filter/output_pin.cpp
===================================================================
--- XIOS/trunk/src/filter/output_pin.cpp	(revision 637)
+++ XIOS/trunk/src/filter/output_pin.cpp	(revision 637)
@@ -0,0 +1,25 @@
+#include "output_pin.hpp"
+#include "exception.hpp"
+
+namespace xios
+{
+  void COutputPin::connectOutput(boost::shared_ptr<CInputPin> inputPin, size_t inputSlot)
+  {
+    if (!inputPin)
+      ERROR("void COutputPin::connectOutput(CInputPin* inputPin, size_t inputSlot)",
+            "The input pin cannot be null.");
+
+    outputs.push_back(std::make_pair(inputPin, inputSlot));
+  }
+
+  void COutputPin::deliverOuput(CDataPacketPtr packet)
+  {
+    if (!packet)
+      ERROR("void COutputPin::deliverOuput(CDataPacketPtr packet)",
+            "The packet cannot be null.");
+
+    std::vector<std::pair<boost::shared_ptr<CInputPin>, size_t> >::iterator it, itEnd;
+    for (it = outputs.begin(), itEnd = outputs.end(); it != itEnd; ++it)
+      it->first->setInput(it->second, packet);
+  }
+} // namespace xios
Index: XIOS/trunk/src/filter/output_pin.hpp
===================================================================
--- XIOS/trunk/src/filter/output_pin.hpp	(revision 637)
+++ XIOS/trunk/src/filter/output_pin.hpp	(revision 637)
@@ -0,0 +1,37 @@
+#ifndef __XIOS_COutputPin__
+#define __XIOS_COutputPin__
+
+#include "input_pin.hpp"
+
+namespace xios
+{
+  /*!
+   * An output pin handles the connections with downstream filters.
+   */
+  class COutputPin
+  {
+    public:
+      /*!
+       * Connects to a specific slot of the input pin of a downstream filter.
+       * Note that the output pin holds a reference on the downstream filter.
+       *
+       * \param inputPin the input pin to connect
+       * \param inputSlot the input slot number
+       */
+      void connectOutput(boost::shared_ptr<CInputPin> inputPin, size_t inputSlot);
+
+    protected:
+      /*!
+       * Delivers an output packet to the downstreams filter.
+       *
+       * \param packet the packet to output
+       */
+      void deliverOuput(CDataPacketPtr packet);
+
+    private:
+      //!< The list of connected filters and the corresponding slot numbers
+      std::vector<std::pair<boost::shared_ptr<CInputPin>, size_t> > outputs;
+  }; // class COutputPin
+} // namespace xios
+
+#endif //__XIOS_COutputPin__
