Add Communication Aggregator Component (#4264)

* Add aggregator component

* Add com aggregator to subtopology

* Fix aggregator issues

* Format and SDD

* Add basic UTs

* Fix not-empty check, test

* sp

* Fix author tag

* Bump GDS for aggregation; timeout aggregator

* Bump comQueue event size

* Increase timeout for integration tests

* Update Fw/Buffer/Buffer.hpp

Co-authored-by: Thomas Boyer-Chammard <49786685+thomas-bc@users.noreply.github.com>

* Update Svc/ComAggregator/CMakeLists.txt

Co-authored-by: Thomas Boyer-Chammard <49786685+thomas-bc@users.noreply.github.com>

* Update Svc/ComAggregator/docs/sdd.md

Co-authored-by: Thomas Boyer-Chammard <49786685+thomas-bc@users.noreply.github.com>

* Update Svc/ComAggregator/docs/sdd.md

Co-authored-by: Thomas Boyer-Chammard <49786685+thomas-bc@users.noreply.github.com>

* Remove unused variable 'good' in action doClear

* A usage note about APID.

---------

Co-authored-by: Thomas Boyer-Chammard <49786685+thomas-bc@users.noreply.github.com>

Fw/Buffer/Buffer.hpp

@@ -47,6 +47,15 @@ namespace Fw {
 class Buffer : public Fw::Serializable {
     friend class Fw::BufferTester;
 
+  public:
+    //! Buffer ownership state
+    //!
+    //! A convenience enumeration to help users implement ownership tracking of buffers.
+    enum class OwnershipState {
+        NOT_OWNED,  //!< The buffer is currently not owned
+        OWNED,      //!< The buffer is currently owned
+    };
+
   public:
     //! The size type for a buffer - for backwards compatibility
     using SizeType = FwSizeType;

Ref/Top/topology.fpp

@@ -88,6 +88,7 @@ module Ref {
       rateGroup1Comp.RateGroupMemberOut[4] -> systemResources.run
       rateGroup1Comp.RateGroupMemberOut[5] -> ComCcsds.comQueue.run
       rateGroup1Comp.RateGroupMemberOut[6] -> CdhCore.cmdDisp.run
+      rateGroup1Comp.RateGroupMemberOut[7] -> ComCcsds.aggregator.timeout
 
       # Rate group 2
       rateGroupDriverComp.CycleOut[Ports_RateGroups.rateGroup2] -> rateGroup2Comp.CycleIn

Svc/CMakeLists.txt

@@ -71,3 +71,4 @@ add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/LinuxTimer/")
 add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Version/")
 add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/FpySequencer/")
 add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Ccsds/")
+add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/ComAggregator/")

Svc/ComAggregator/CMakeLists.txt

@@ -0,0 +1,34 @@
+####
+# F Prime CMakeLists.txt:
+#
+# SOURCES: list of source files (to be compiled)
+# AUTOCODER_INPUTS: list of files to be passed to the autocoders
+# DEPENDS: list of libraries that this module depends on
+#
+# More information in the F´ CMake API documentation:
+# https://fprime.jpl.nasa.gov/latest/docs/reference/api/cmake/API/
+#
+####
+
+# Module names are derived from the path from the nearest project/library/framework
+# root when not specifically overridden by the developer. i.e. The module defined by
+# `Ref/SignalGen/CMakeLists.txt` will be named `Ref_SignalGen`.
+
+register_fprime_library(
+    AUTOCODER_INPUTS
+        "${CMAKE_CURRENT_LIST_DIR}/ComAggregator.fpp"
+    SOURCES
+        "${CMAKE_CURRENT_LIST_DIR}/ComAggregator.cpp"
+)
+
+### Unit Tests ###
+register_fprime_ut(
+    AUTOCODER_INPUTS
+        "${CMAKE_CURRENT_LIST_DIR}/ComAggregator.fpp"
+    SOURCES
+        "${CMAKE_CURRENT_LIST_DIR}/test/ut/ComAggregatorTestMain.cpp"
+        "${CMAKE_CURRENT_LIST_DIR}/test/ut/ComAggregatorTester.cpp"
+    DEPENDS
+        STest # For rules-based testing
+    UT_AUTO_HELPERS
+)

Svc/ComAggregator/ComAggregator.cpp

@@ -0,0 +1,120 @@
+// ======================================================================
+// \title  ComAggregator.cpp
+// \author lestarch
+// \brief  cpp file for ComAggregator component implementation class
+// ======================================================================
+
+#include "Svc/ComAggregator/ComAggregator.hpp"
+
+namespace Svc {
+
+// ----------------------------------------------------------------------
+// Component construction and destruction
+// ----------------------------------------------------------------------
+
+ComAggregator ::ComAggregator(const char* const compName)
+    : ComAggregatorComponentBase(compName),
+      m_bufferState(Fw::Buffer::OwnershipState::OWNED),
+      m_frameBuffer(m_frameBufferStore, sizeof(m_frameBufferStore)),
+      m_frameSerializer(m_frameBuffer.getSerializer()) {}
+
+ComAggregator ::~ComAggregator() {}
+
+void ComAggregator ::preamble() {
+    Fw::Success good = Fw::Success::SUCCESS;
+    this->comStatusOut_out(0, good);
+}
+
+// ----------------------------------------------------------------------
+// Handler implementations for typed input ports
+// ----------------------------------------------------------------------
+
+void ComAggregator ::comStatusIn_handler(FwIndexType portNum, Fw::Success& condition) {
+    this->aggregationMachine_sendSignal_status(condition);
+}
+
+void ComAggregator ::dataIn_handler(FwIndexType portNum, Fw::Buffer& data, const ComCfg::FrameContext& context) {
+    Svc::ComDataContextPair pair(data, context);
+    this->aggregationMachine_sendSignal_fill(pair);
+}
+
+void ComAggregator ::dataReturnIn_handler(FwIndexType portNum, Fw::Buffer& data, const ComCfg::FrameContext& context) {
+    FW_ASSERT(this->m_bufferState == Fw::Buffer::OwnershipState::NOT_OWNED);
+    this->m_bufferState = Fw::Buffer::OwnershipState::OWNED;
+}
+
+void ComAggregator ::timeout_handler(FwIndexType portNum, U32 context) {
+    this->aggregationMachine_sendSignal_timeout();
+}
+
+// ----------------------------------------------------------------------
+// Implementations for internal state machine actions
+// ----------------------------------------------------------------------
+
+void ComAggregator ::Svc_AggregationMachine_action_doClear(SmId smId, Svc_AggregationMachine::Signal signal) {
+    this->m_frameSerializer.resetSer();
+    this->m_frameBuffer.setSize(sizeof(this->m_frameBufferStore));
+    if (this->m_held.get_data().isValid()) {
+        // Fill the held data
+        this->Svc_AggregationMachine_action_doFill(smId, signal, this->m_held);
+        this->m_held = Svc::ComDataContextPair();
+    }
+}
+
+void ComAggregator ::Svc_AggregationMachine_action_doFill(SmId smId,
+                                                          Svc_AggregationMachine::Signal signal,
+                                                          const Svc::ComDataContextPair& value) {
+    Fw::SerializeStatus status = this->m_frameSerializer.serializeFrom(
+        value.get_data().getData(), value.get_data().getSize(), Fw::Serialization::OMIT_LENGTH);
+    FW_ASSERT(status == Fw::SerializeStatus::FW_SERIALIZE_OK);
+    this->m_lastContext = value.get_context();
+    Fw::Success good = Fw::Success::SUCCESS;
+    // Return port does not alter data and thus const-cast is safe
+    this->dataReturnOut_out(0, const_cast<Fw::Buffer&>(value.get_data()), value.get_context());
+    this->comStatusOut_out(0, good);
+}
+
+void ComAggregator ::Svc_AggregationMachine_action_doSend(SmId smId, Svc_AggregationMachine::Signal signal) {
+    // Send only when the buffer will be valid
+    if (this->m_frameSerializer.getBuffLength() > 0) {
+        this->m_bufferState = Fw::Buffer::OwnershipState::NOT_OWNED;
+        this->m_frameBuffer.setSize(this->m_frameSerializer.getBuffLength());
+        this->dataOut_out(0, this->m_frameBuffer, this->m_lastContext);
+    }
+}
+
+void ComAggregator ::Svc_AggregationMachine_action_doHold(SmId smId,
+                                                          Svc_AggregationMachine::Signal signal,
+                                                          const Svc::ComDataContextPair& value) {
+    FW_ASSERT(not this->m_held.get_data().isValid());
+    this->m_held = value;
+}
+
+void ComAggregator ::Svc_AggregationMachine_action_assertNoStatus(SmId smId, Svc_AggregationMachine::Signal signal) {
+    // Status is not possible in this state, confirm by assertion
+    FW_ASSERT(0);
+}
+
+// ----------------------------------------------------------------------
+// Implementations for internal state machine guards
+// ----------------------------------------------------------------------
+
+bool ComAggregator ::Svc_AggregationMachine_guard_isFull(SmId smId,
+                                                         Svc_AggregationMachine::Signal signal,
+                                                         const Svc::ComDataContextPair& value) const {
+    FW_ASSERT(value.get_data().getSize() <= ComCfg::AggregationSize);
+    const FwSizeType remaining = this->m_frameSerializer.getBuffCapacity() - this->m_frameSerializer.getBuffLength();
+    return (remaining <= value.get_data().getSize());
+}
+
+bool ComAggregator ::Svc_AggregationMachine_guard_isNotEmpty(SmId smId, Svc_AggregationMachine::Signal signal) const {
+    return this->m_frameSerializer.getBuffLength() > 0;
+}
+
+bool ComAggregator ::Svc_AggregationMachine_guard_isGood(SmId smId,
+                                                         Svc_AggregationMachine::Signal signal,
+                                                         const Fw::Success& value) const {
+    return value == Fw::Success::SUCCESS;
+}
+
+}  // namespace Svc

Svc/ComAggregator/ComAggregator.fpp

@@ -0,0 +1,82 @@
+module Svc {
+    @ Aggregation state machine
+    state machine AggregationMachine {
+        @ Initial state: WAIT_STATUS
+        @ Wait for initial 'status' to start the process
+        initial enter WAIT_STATUS
+
+        @ Rate-group driven timeout signal
+        signal timeout
+
+        @ Fill buffer signal
+        signal fill: Svc.ComDataContextPair
+
+        @ Status return
+        signal status: Fw.Success
+
+        @ Check if full 
+        guard isFull: Svc.ComDataContextPair
+
+        @ Check if not empty
+        guard isNotEmpty
+
+        @ Check if last status is good 
+        guard isGood: Fw.Success
+
+        @ Clear the buffer fill state, last status
+        action doClear
+
+        @ Fill the buffer with data
+        action doFill: Svc.ComDataContextPair
+
+        @ Send the buffer data
+        action doSend
+
+        @ Hold a buffer
+        action doHold: Svc.ComDataContextPair
+
+
+        @ Assert no status when in fill state
+        action assertNoStatus
+
+        @ The IS_FULL_THEN_SEND choice
+        choice IS_FULL_THEN_SEND {
+            if isFull do { doHold, doSend } enter WAIT_STATUS \
+                else do { doFill } enter FILL
+        }
+
+        @ The IS_GOOD_STATUS choice
+        choice IS_GOOD_STATUS {
+            if isGood do { doClear } enter FILL \
+                else enter WAIT_STATUS
+        }
+        
+        @ Wait for com status from downstream
+        state WAIT_STATUS {
+            # ASSERT: fill cannot happen before initial 'status'
+            on fill do { doHold }
+            # IGNORE: 'timeout', this signal is irrelevant
+            # On status, move to IS_GOOD_STATUS choice
+            on status enter IS_GOOD_STATUS
+        }
+
+        @ Buffer aggregation in-progress
+        state FILL {
+            # Fill buffer, check if full then send or fill
+            on fill enter IS_FULL_THEN_SEND
+            # Timeout, send buffer
+            on timeout if isNotEmpty do { doSend } enter WAIT_STATUS
+            # ASSERT: status cannot happen while filling
+            on status do { assertNoStatus }
+        }
+    }
+
+    @ Aggregates com buffers
+    active component ComAggregator {
+        import Svc.Framer
+        sync input port timeout: Svc.Sched
+
+        @ State machine instance for aggregation state machine
+        state machine instance aggregationMachine: AggregationMachine
+    }
+}

Svc/ComAggregator/ComAggregator.hpp

@@ -0,0 +1,146 @@
+// ======================================================================
+// \title  ComAggregator.hpp
+// \author lestarch
+// \brief  hpp file for ComAggregator component implementation class
+// ======================================================================
+
+#ifndef Svc_ComAggregator_HPP
+#define Svc_ComAggregator_HPP
+
+#include "Os/Mutex.hpp"
+#include "Svc/ComAggregator/ComAggregatorComponentAc.hpp"
+
+namespace Svc {
+
+class ComAggregator final : public ComAggregatorComponentBase {
+    friend class ComAggregatorTester;  // Allow unit test access to private members
+  public:
+    // ----------------------------------------------------------------------
+    // Component construction and destruction
+    // ----------------------------------------------------------------------
+
+    //! Construct ComAggregator object
+    ComAggregator(const char* const compName  //!< The component name
+    );
+
+    //! Destroy ComAggregator object
+    ~ComAggregator();
+
+    void preamble() override;
+
+  private:
+    // ----------------------------------------------------------------------
+    // Handler implementations for typed input ports
+    // ----------------------------------------------------------------------
+
+    //! Handler implementation for comStatusIn
+    //!
+    //! Port receiving the general status from the downstream component
+    //! indicating it is ready or not-ready for more input
+    void comStatusIn_handler(FwIndexType portNum,    //!< The port number
+                             Fw::Success& condition  //!< Condition success/failure
+                             ) override;
+
+    //! Handler implementation for dataIn
+    //!
+    //! Port to receive data to frame, in a Fw::Buffer with optional context
+    void dataIn_handler(FwIndexType portNum,  //!< The port number
+                        Fw::Buffer& data,
+                        const ComCfg::FrameContext& context) override;
+
+    //! Handler implementation for dataReturnIn
+    //!
+    //! Buffer coming from a deallocate call in a ComDriver component
+    void dataReturnIn_handler(FwIndexType portNum,  //!< The port number
+                              Fw::Buffer& data,
+                              const ComCfg::FrameContext& context) override;
+
+    //! Handler implementation for timeout
+    void timeout_handler(FwIndexType portNum,  //!< The port number
+                         U32 context           //!< The call order
+                         ) override;
+
+  private:
+    // ----------------------------------------------------------------------
+    // Implementations for internal state machine actions
+    // ----------------------------------------------------------------------
+
+    //! Implementation for action doClear of state machine Svc_AggregationMachine
+    //!
+    //! Clear the buffer fill state, last status
+    void Svc_AggregationMachine_action_doClear(SmId smId,                             //!< The state machine id
+                                               Svc_AggregationMachine::Signal signal  //!< The signal
+                                               ) override;
+
+    //! Implementation for action doFill of state machine Svc_AggregationMachine
+    //!
+    //! Fill the buffer with data
+    void Svc_AggregationMachine_action_doFill(SmId smId,                              //!< The state machine id
+                                              Svc_AggregationMachine::Signal signal,  //!< The signal
+                                              const Svc::ComDataContextPair& value    //!< The value
+                                              ) override;
+
+    //! Implementation for action doSend of state machine Svc_AggregationMachine
+    //!
+    //! Send the buffer data
+    void Svc_AggregationMachine_action_doSend(SmId smId,                             //!< The state machine id
+                                              Svc_AggregationMachine::Signal signal  //!< The signal
+                                              ) override;
+
+    //! Implementation for action doHold of state machine Svc_AggregationMachine
+    //!
+    //! Hold a buffer
+    void Svc_AggregationMachine_action_doHold(SmId smId,                              //!< The state machine id
+                                              Svc_AggregationMachine::Signal signal,  //!< The signal
+                                              const Svc::ComDataContextPair& value    //!< The value
+                                              ) override;
+
+    //! Implementation for action assertNoStatus of state machine Svc_AggregationMachine
+    //!
+    //! Assert no status when in fill state
+    void Svc_AggregationMachine_action_assertNoStatus(SmId smId,                             //!< The state machine id
+                                                      Svc_AggregationMachine::Signal signal  //!< The signal
+                                                      ) override;
+
+  private:
+    // ----------------------------------------------------------------------
+    // Implementations for internal state machine guards
+    // ----------------------------------------------------------------------
+
+    //! Implementation for guard isFull of state machine Svc_AggregationMachine
+    //!
+    //! Check if full
+    bool Svc_AggregationMachine_guard_isFull(SmId smId,                              //!< The state machine id
+                                             Svc_AggregationMachine::Signal signal,  //!< The signal
+                                             const Svc::ComDataContextPair& value    //!< The value
+    ) const override;
+
+    //! Implementation for guard isNotEmpty of state machine Svc_AggregationMachine
+    //!
+    //! Check if not empty
+    bool Svc_AggregationMachine_guard_isNotEmpty(SmId smId,                             //!< The state machine id
+                                                 Svc_AggregationMachine::Signal signal  //!< The signal
+    ) const override;
+
+    //! Implementation for guard isGood of state machine Svc_AggregationMachine
+    //!
+    //! Check if last status is good
+    bool Svc_AggregationMachine_guard_isGood(SmId smId,                              //!< The state machine id
+                                             Svc_AggregationMachine::Signal signal,  //!< The signal
+                                             const Fw::Success& value                //!< The value
+    ) const override;
+
+  private:
+    U8 m_frameBufferStore[ComCfg::AggregationSize];  //!< Buffer to hold the frame data
+    Fw::Buffer::OwnershipState m_bufferState =
+        Fw::Buffer::OwnershipState::OWNED;  //!< whether m_frameBuffer is owned by TmFramer
+    Fw::Buffer m_frameBuffer;
+    Fw::ExternalSerializeBufferWithMemberCopy m_frameSerializer;  //!< Serializer for m_frameBuffer
+    ComCfg::FrameContext m_lastContext;                           //!< Context for the current frame
+
+    Svc::ComDataContextPair m_held;  //!< Held data while waiting for send
+};
+
+}  // namespace Svc
+
+#endif

Svc/ComAggregator/docs/img/diagram.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" id="fppDiagrams_0_root" class="sprotty-graph" style="border-block-end-color:rgb(204, 204, 204);border-block-start-color:rgb(204, 204, 204);border-bottom-color:rgb(204, 204, 204);border-inline-end-color:rgb(204, 204, 204);border-inline-start-color:rgb(204, 204, 204);border-left-color:rgb(204, 204, 204);border-right-color:rgb(204, 204, 204);border-top-color:rgb(204, 204, 204);caret-color:rgb(204, 204, 204);color:rgb(204, 204, 204);column-rule-color:rgb(204, 204, 204);flex-basis:0%;flex-grow:1;font-family:&quot;Helvetica Neue&quot;, Helvetica, Arial, sans-serif;font-size:13px;outline-color:rgb(204, 204, 204);perspective-origin:0px 75px;scrollbar-color:rgba(121, 121, 121, 0.4) rgb(31, 31, 31);text-decoration:none solid rgb(204, 204, 204);text-decoration-color:rgb(204, 204, 204);text-emphasis-color:rgb(204, 204, 204);transform-origin:0px 75px;-webkit-locale:&quot;en&quot;;-webkit-text-fill-color:rgb(204, 204, 204);-webkit-text-stroke-color:rgb(204, 204, 204);" version="1.1" viewBox="12.000007629394531 12 284.5657653808594 90" width="284.5657653808594" height="90"><g transform="scale(1) translate(0,0)"><g id="fppDiagrams_0_uninstantiatedComponent" class="component" transform="translate(98.13125610351562, 12)"><rect rx="10" class="sprotty-node task component-active" width="102.19701385498047" height="90" style="fill:rgb(0, 120, 212);stroke:rgb(63, 185, 132);stroke-width:3px;"/><text class="node:component sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.label.0" style="fill:rgb(204, 204, 204);"/><text class="node:component sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.label.1" transform="translate(5, 37.799999713897705) translate(0, 12)" style="fill:rgb(204, 204, 204);">ComAggregator</text><g id="fppDiagrams_0_uninstantiatedComponent.dataIn.0" transform="translate(-10, 30)"><rect width="10" height="10" class="sprotty-port port-sync" style="fill:rgb(254, 146, 168);stroke:rgb(123, 51, 125);"/><text class="port sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.dataIn.0.label.0" transform="translate(-37.13125228881836, -2.200000286102295) translate(0, 12)" style="fill:rgb(204, 204, 204);">dataIn</text></g><g id="fppDiagrams_0_uninstantiatedComponent.dataOut.0" transform="translate(102.19701385498047, 15)"><rect width="10" height="10" class="sprotty-port" style="fill:rgb(204, 204, 204);stroke:rgb(123, 51, 125);"/><text class="port sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.dataOut.0.label.0" transform="translate(11, -2.200000286102295) translate(0, 12)" style="fill:rgb(204, 204, 204);">dataOut</text></g><g id="fppDiagrams_0_uninstantiatedComponent.dataReturnOut.0" transform="translate(102.19701385498047, 40)"><rect width="10" height="10" class="sprotty-port" style="fill:rgb(204, 204, 204);stroke:rgb(123, 51, 125);"/><text class="port sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.dataReturnOut.0.label.0" transform="translate(11, -2.200000286102295) translate(0, 12)" style="fill:rgb(204, 204, 204);">dataReturnOut</text></g><g id="fppDiagrams_0_uninstantiatedComponent.dataReturnIn.0" transform="translate(-10, 50)"><rect width="10" height="10" class="sprotty-port port-sync" style="fill:rgb(254, 146, 168);stroke:rgb(123, 51, 125);"/><text class="port sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.dataReturnIn.0.label.0" transform="translate(-76.13125610351562, -2.200000286102295) translate(0, 12)" style="fill:rgb(204, 204, 204);">dataReturnIn</text></g><g id="fppDiagrams_0_uninstantiatedComponent.comStatusIn.0" transform="translate(-10, 70)"><rect width="10" height="10" class="sprotty-port port-sync" style="fill:rgb(254, 146, 168);stroke:rgb(123, 51, 125);"/><text class="port sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.comStatusIn.0.label.0" transform="translate(-73.22500610351562, -2.200000286102295) translate(0, 12)" style="fill:rgb(204, 204, 204);">comStatusIn</text></g><g id="fppDiagrams_0_uninstantiatedComponent.comStatusOut.0" transform="translate(102.19701385498047, 65)"><rect width="10" height="10" class="sprotty-port" style="fill:rgb(204, 204, 204);stroke:rgb(123, 51, 125);"/><text class="port sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.comStatusOut.0.label.0" transform="translate(11, -2.200000286102295) translate(0, 12)" style="fill:rgb(204, 204, 204);">comStatusOut</text></g><g id="fppDiagrams_0_uninstantiatedComponent.timeout.0" transform="translate(-10, 10)"><rect width="10" height="10" class="sprotty-port port-sync" style="fill:rgb(254, 146, 168);stroke:rgb(123, 51, 125);"/><text class="port sprotty-label" id="fppDiagrams_0_uninstantiatedComponent.timeout.0.label.0" transform="translate(-43.618751525878906, -2.200000286102295) translate(0, 12)" style="fill:rgb(204, 204, 204);">timeout</text></g></g></g></svg>

Svc/ComAggregator/docs/sdd.md

@@ -0,0 +1,25 @@
+# Svc::ComAggregator
+
+Aggregates buffers in the downlink chain. This is for use with systems that have fixed size frames (e.g. CCSDS TM) that needed internal aggregation.
+
+> [!CAUTION]
+> `Svc::ComAggregator` does not preserve context.
+
+## Requirements
+
+| ID                    | Description                                                                                                                                                   | Verification |
+| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------ |
+| Svc-ComAggregator-001 | ComAggregator shall accept incoming downlink data as `Fw::Buffer`, `ComCfg::FrameContext` pairs and append the buffer into the aggregate space permitting     | Unit-Test    |
+| Svc-ComAggregator-002 | ComAggregator shall hold the incoming buffer when there is insufficient space in the aggregate buffer.                                                        | Unit-Test    |
+| Svc-ComAggregator-003 | ComAggregator shall send the current aggregate buffer when the incoming buffer is held due to overflow.                                                       | Unit-Test    |
+| Svc-ComAggregator-004 | ComAggregator shall send the current aggregate buffer when it receives a timeout trigger if and only if the aggregate is non-empty.                           | Unit-Test    |
+| Svc-ComAggregator-005 | ComAggregator shall clear aggregation state when a SUCCESS communication status is received back.                                                             | Unit-Test    |
+| Svc-ComAggregator-006 | ComAggregator shall preserve the order of received buffers when forming each aggregate and across aggregate sends.                                            | Unit-Test    |
+| Svc-ComAggregator-007 | ComAggregator shall inter operate with the [Communication Adapter Interface comStatus protocol](../../../docs/reference/communication-adapter-interface.md)   | Unit-Test    |
+
+
+## Design
+
+![Component Block Diagram](./img/diagram.svg)
+
+`Svc.ComAggregator` implements `Svc.Framer`.  Additionally, it has a `Svc.Sched` timeout port enabling timeout to be driven via a rate group.

Svc/ComAggregator/test/ut/ComAggregatorTestMain.cpp

@@ -0,0 +1,69 @@
+// ======================================================================
+// \title  ComAggregatorTestMain.cpp
+// \author lestarch
+// \brief  cpp file for ComAggregator component test main function
+// ======================================================================
+
+#include "ComAggregatorTester.hpp"
+
+TEST(Nominal, Initial) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+}
+
+TEST(Nominal, Fill) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+    tester.test_fill();
+}
+
+TEST(Nominal, MultiFill) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+    tester.test_fill_multi();
+}
+
+TEST(Nominal, Full) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+    tester.test_fill_multi();
+    tester.test_full();
+}
+
+TEST(Nominal, Timeout) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+    tester.test_fill_multi();
+    tester.test_timeout();
+}
+
+TEST(OffNominal, TimeoutEmpty) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+    tester.test_timeout_zero();
+    tester.test_fill_multi();
+    tester.test_full();
+}
+
+TEST(Nominal, HoldWhileWaiting) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+    tester.test_fill_multi();
+    tester.test_hold_while_waiting();
+}
+
+TEST(Nominal, Clear) {
+    Svc::ComAggregatorTester tester;
+    tester.test_initial();
+    tester.test_fill_multi();
+    tester.test_full();
+    tester.test_fill_multi();
+    tester.test_timeout();
+    tester.test_fill_multi();
+    tester.test_full();
+}
+
+int main(int argc, char** argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

Svc/ComAggregator/test/ut/ComAggregatorTester.cpp

@@ -0,0 +1,225 @@
+// ======================================================================
+// \title  ComAggregatorTester.cpp
+// \author lestarch
+// \brief  cpp file for ComAggregator component test harness implementation class
+// ======================================================================
+
+#include "ComAggregatorTester.hpp"
+#include <STest/Pick/Pick.hpp>
+#include <vector>
+#include "config/FppConstantsAc.hpp"
+
+namespace Svc {
+
+// ----------------------------------------------------------------------
+// Construction and destruction
+// ----------------------------------------------------------------------
+
+ComAggregatorTester ::ComAggregatorTester()
+    : ComAggregatorGTestBase("ComAggregatorTester", ComAggregatorTester::MAX_HISTORY_SIZE), component("ComAggregator") {
+    this->initComponents();
+    this->connectPorts();
+}
+
+ComAggregatorTester ::~ComAggregatorTester() {}
+
+// ----------------------------------------------------------------------
+// Tests
+// ----------------------------------------------------------------------
+
+Fw::Buffer ComAggregatorTester ::fill_buffer(U32 size) {
+    EXPECT_GT(size, 0);
+    U8* data = new U8[size];
+    for (U32 i = 0; i < size; i++) {
+        data[i] = static_cast<U8>(STest::Pick::lowerUpper(0, 255));
+    }
+    Fw::Buffer buffer(data, size);
+    return buffer;
+}
+
+//! Shadow aggregate a buffer for validation
+void ComAggregatorTester ::shadow_aggregate(const Fw::Buffer& buffer) {
+    for (FwSizeType i = 0; i < buffer.getSize(); i++) {
+        this->m_aggregation.push_back(buffer.getData()[i]);
+    }
+}
+
+//! Validate against shadow aggregation
+void ComAggregatorTester ::validate_aggregation(const Fw::Buffer& buffer) {
+    ASSERT_EQ(buffer.getSize(), this->m_aggregation.size());
+    for (FwSizeType i = 0; i < this->m_aggregation.size(); i++) {
+        ASSERT_EQ(buffer.getData()[i], this->m_aggregation[i]);
+    }
+}
+
+void ComAggregatorTester ::validate_buffer_aggregated(const Fw::Buffer& buffer, const ComCfg::FrameContext& context) {
+    FwSizeType start = this->component.m_frameSerializer.getBuffLength() - buffer.getSize();
+    for (FwSizeType i = 0; i < buffer.getSize(); i++) {
+        ASSERT_EQ(buffer.getData()[i], this->component.m_frameBuffer.getData()[start + i]);
+    }
+    ASSERT_EQ(context, this->component.m_lastContext);
+    this->shadow_aggregate(buffer);
+    delete[] buffer.getData();
+}
+
+void ComAggregatorTester ::test_initial() {
+    // Initial state should have empty buffer
+    ASSERT_EQ(this->component.m_frameSerializer.getBuffLength(), 0);
+    ASSERT_EQ(this->component.m_bufferState, Fw::Buffer::OwnershipState::OWNED);
+    this->component.preamble();
+    ASSERT_from_comStatusOut(0, Fw::Success::SUCCESS);
+    Fw::Success good = Fw::Success::SUCCESS;
+    this->invoke_to_comStatusIn(0, good);
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+}
+
+//! Tests fill operation
+Fw::Buffer ComAggregatorTester ::test_fill(bool expect_hold) {
+    // Precondition: initial has run
+    const FwSizeType ORIGINAL_LENGTH = this->component.m_frameSerializer.getBuffLength();
+    if (ORIGINAL_LENGTH == ComCfg::AggregationSize) {
+        // Nothing to fill
+        return Fw::Buffer();
+    }
+    const U32 BUFFER_LENGTH = STest::Pick::lowerUpper(1, static_cast<U32>(ComCfg::AggregationSize - ORIGINAL_LENGTH));
+    Fw::Buffer buffer = fill_buffer(BUFFER_LENGTH);
+    ComCfg::FrameContext context;
+
+    this->invoke_to_dataIn(0, buffer, context);
+    EXPECT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    if (expect_hold) {
+        EXPECT_EQ(ORIGINAL_LENGTH, this->component.m_frameSerializer.getBuffLength());
+    } else {
+        EXPECT_EQ(ORIGINAL_LENGTH + BUFFER_LENGTH, this->component.m_frameSerializer.getBuffLength());
+        this->validate_buffer_aggregated(buffer, context);
+    }
+    this->clearHistory();
+    return buffer;
+}
+
+void ComAggregatorTester ::test_fill_multi() {
+    U32 count = STest::Pick::lowerUpper(1, 5);
+    for (U32 i = 0; i < count; i++) {
+        (void)this->test_fill();
+    }
+}
+
+//! Tests full operation
+void ComAggregatorTester ::test_full() {
+    // Precondition: fill has run
+    // Chose a buffer that will be too large to fit but still will fit after being aggregated
+    const FwSizeType ORIGINAL_LENGTH = this->component.m_frameSerializer.getBuffLength();
+    const U32 BUFFER_LENGTH = STest::Pick::lowerUpper(static_cast<U32>(ComCfg::AggregationSize - ORIGINAL_LENGTH + 1),
+                                                      static_cast<U32>(ComCfg::AggregationSize));
+    Fw::Buffer buffer = fill_buffer(BUFFER_LENGTH);
+    ComCfg::FrameContext context;
+
+    // Send the overflow buffer and ensure the current aggregation comes out
+    this->invoke_to_dataIn(0, buffer, context);
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    ASSERT_from_dataOut_SIZE(1);
+    this->validate_aggregation(this->fromPortHistory_dataOut->at(0).data);
+
+    // Invoke some number of failures
+    for (U32 i = 0; i < STest::Pick::lowerUpper(1, 5); i++) {
+        Fw::Success bad = Fw::Success::FAILURE;
+        this->invoke_to_comStatusIn(0, bad);
+        ASSERT_EQ(this->dispatchOne(this->component),
+                  Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+        // Should be no change
+        this->validate_aggregation(this->component.m_frameBuffer);
+        ASSERT_from_dataOut_SIZE(1);
+    }
+    // Const cast is safe as data is not altered
+    this->invoke_to_dataReturnIn(0, const_cast<Fw::Buffer&>(this->fromPortHistory_dataOut->at(0).data),
+                                 this->fromPortHistory_dataOut->at(0).context);
+    Fw::Success good = Fw::Success::SUCCESS;
+    this->invoke_to_comStatusIn(0, good);
+    this->m_aggregation.clear();
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    // Validate that the new buffer has been aggregated
+    this->validate_buffer_aggregated(buffer, context);
+    this->clearHistory();
+}
+
+//! Tests timeout operation
+void ComAggregatorTester ::test_timeout() {
+    // Precondition: fill has run
+    this->invoke_to_timeout(0, 0);
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    ASSERT_from_dataOut_SIZE(1);
+    this->validate_aggregation(this->fromPortHistory_dataOut->at(0).data);
+
+    // Invoke some number of failures
+    for (U32 i = 0; i < STest::Pick::lowerUpper(1, 5); i++) {
+        Fw::Success bad = Fw::Success::FAILURE;
+        this->invoke_to_comStatusIn(0, bad);
+        ASSERT_EQ(this->dispatchOne(this->component),
+                  Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+        // Should be no change
+        this->validate_aggregation(this->component.m_frameBuffer);
+        ASSERT_from_dataOut_SIZE(1);
+    }
+    // Const cast is safe as data is not altered
+    this->invoke_to_dataReturnIn(0, const_cast<Fw::Buffer&>(this->fromPortHistory_dataOut->at(0).data),
+                                 this->fromPortHistory_dataOut->at(0).context);
+    Fw::Success good = Fw::Success::SUCCESS;
+    this->invoke_to_comStatusIn(0, good);
+    this->m_aggregation.clear();
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    this->clearHistory();
+}
+
+void ComAggregatorTester ::test_timeout_zero() {
+    // Precondition: initialize has run
+    this->invoke_to_timeout(0, 0);
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    ASSERT_from_dataOut_SIZE(0);
+    this->clearHistory();
+}
+
+//! Tests hold while waiting on data return
+void ComAggregatorTester ::test_hold_while_waiting() {
+    // Precondition: fill has run
+    ComCfg::FrameContext context;
+    this->invoke_to_timeout(0, 0);
+
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    ASSERT_from_dataOut_SIZE(1);
+    this->validate_aggregation(this->fromPortHistory_dataOut->at(0).data);
+    Fw::Buffer major_buffer = this->fromPortHistory_dataOut->at(0).data;
+
+    // Invoke some number of failures
+    for (U32 i = 0; i < STest::Pick::lowerUpper(1, 5); i++) {
+        Fw::Success bad = Fw::Success::FAILURE;
+        this->invoke_to_comStatusIn(0, bad);
+        ASSERT_EQ(this->dispatchOne(this->component),
+                  Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+        // Should be no change
+        this->validate_aggregation(this->component.m_frameBuffer);
+        ASSERT_from_dataOut_SIZE(1);
+    }
+    // Force a hold
+    Fw::Buffer minor_buffer = this->test_fill(true);
+
+    // Const cast is safe as data is not altered
+    this->invoke_to_dataReturnIn(0, major_buffer, context);
+    Fw::Success good = Fw::Success::SUCCESS;
+    this->invoke_to_comStatusIn(0, good);
+    this->m_aggregation.clear();
+    ASSERT_EQ(this->dispatchOne(this->component),
+              Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK);  // Dispatch the state machine
+    // Validate that the new buffer has been aggregated
+    this->validate_buffer_aggregated(minor_buffer, context);
+    this->clearHistory();
+}
+
+}  // namespace Svc

Svc/ComAggregator/test/ut/ComAggregatorTester.hpp

@@ -0,0 +1,110 @@
+// ======================================================================
+// \title  ComAggregatorTester.hpp
+// \author lestarch
+// \brief  hpp file for ComAggregator component test harness implementation class
+// ======================================================================
+
+#ifndef Svc_ComAggregatorTester_HPP
+#define Svc_ComAggregatorTester_HPP
+
+#include <vector>
+#include "Svc/ComAggregator/ComAggregator.hpp"
+#include "Svc/ComAggregator/ComAggregatorGTestBase.hpp"
+
+namespace Svc {
+
+class ComAggregatorTester final : public ComAggregatorGTestBase {
+  public:
+    // ----------------------------------------------------------------------
+    // Constants
+    // ----------------------------------------------------------------------
+
+    // Maximum size of histories storing events, telemetry, and port outputs
+    static const FwSizeType MAX_HISTORY_SIZE = 20;
+
+    // Instance ID supplied to the component instance under test
+    static const FwEnumStoreType TEST_INSTANCE_ID = 0;
+
+    // Queue depth supplied to the component instance under test
+    static const FwSizeType TEST_INSTANCE_QUEUE_DEPTH = 20;
+
+  public:
+    // ----------------------------------------------------------------------
+    // Construction and destruction
+    // ----------------------------------------------------------------------
+
+    //! Construct object ComAggregatorTester
+    ComAggregatorTester();
+
+    //! Destroy object ComAggregatorTester
+    ~ComAggregatorTester();
+
+  public:
+    // ----------------------------------------------------------------------
+    // Tests
+    // ----------------------------------------------------------------------
+
+    //! Tests initial operation
+    void test_initial();
+
+    //! Tests fill operation
+    Fw::Buffer test_fill(bool expect_hold = false);
+
+    //! Tests fill operation
+    void test_fill_multi();
+
+    //! Tests full operation
+    void test_full();
+
+    //! Tests timeout operation
+    void test_timeout();
+
+    //! Tests timeout operation sends no empty buffer
+    void test_timeout_zero();
+
+    //! Tests hold while waiting on data return
+    void test_hold_while_waiting();
+
+    //! Tests clear operation
+    void test_clear();
+
+    //! Tests clear operation with held data
+    void test_clear_with_hold();
+
+    //! Helper to fill a buffer with random data
+    Fw::Buffer fill_buffer(U32 size);
+
+    //! Shadow aggregate a buffer for validation
+    void shadow_aggregate(const Fw::Buffer& buffer);
+
+    //! Validate against shadow aggregation
+    void validate_aggregation(const Fw::Buffer& buffer);
+
+    //! Helper to validate a buffer has been aggregated correctly
+    void validate_buffer_aggregated(const Fw::Buffer& buffer, const ComCfg::FrameContext& context);
+
+  private:
+    // ----------------------------------------------------------------------
+    // Helper functions
+    // ----------------------------------------------------------------------
+
+    //! Connect ports
+    void connectPorts();
+
+    //! Initialize components
+    void initComponents();
+
+  private:
+    // ----------------------------------------------------------------------
+    // Member variables
+    // ----------------------------------------------------------------------
+
+    //! The component under test
+    ComAggregator component;
+    //! Shadow aggregation for validation
+    std::vector<U8> m_aggregation;
+};
+
+}  // namespace Svc
+
+#endif

Svc/Ports/CommsPorts/CommsPorts.fpp

@@ -6,6 +6,12 @@
 #####
 
 module Svc {
+    @ Struct representing a communications data buffer along with context information
+    @ for use storing the inputs ComDataWithContext port
+    struct ComDataContextPair {
+        data: Fw.Buffer
+        context: ComCfg.FrameContext
+    }
  
     @ Port for sending communications data (frames) buffer along with context information
     port ComDataWithContext(ref data: Fw.Buffer, context: ComCfg.FrameContext)

Svc/Subtopologies/ComCcsds/ComCcsds.fpp

@@ -95,14 +95,19 @@ module ComCcsds {
     instance tcDeframer: Svc.Ccsds.TcDeframer base id ComCcsdsConfig.BASE_ID + 0x04000
 
     instance spacePacketDeframer: Svc.Ccsds.SpacePacketDeframer base id ComCcsdsConfig.BASE_ID + 0x05000
+
+    instance aggregator: Svc.ComAggregator base id ComCcsdsConfig.BASE_ID + 0x06000 \
+        queue size ComCcsdsConfig.QueueSizes.aggregator \
+        stack size ComCcsdsConfig.StackSizes.aggregator
+
     # NOTE: name 'framer' is used for the framer that connects to the Com Adapter Interface for better subtopology interoperability
-    instance framer: Svc.Ccsds.TmFramer base id ComCcsdsConfig.BASE_ID + 0x06000
+    instance framer: Svc.Ccsds.TmFramer base id ComCcsdsConfig.BASE_ID + 0x07000
 
-    instance spacePacketFramer: Svc.Ccsds.SpacePacketFramer base id ComCcsdsConfig.BASE_ID + 0x07000
+    instance spacePacketFramer: Svc.Ccsds.SpacePacketFramer base id ComCcsdsConfig.BASE_ID + 0x08000
 
-    instance apidManager: Svc.Ccsds.ApidManager base id ComCcsdsConfig.BASE_ID + 0x08000
+    instance apidManager: Svc.Ccsds.ApidManager base id ComCcsdsConfig.BASE_ID + 0x09000
 
-    instance comStub: Svc.ComStub base id ComCcsdsConfig.BASE_ID + 0x09000
+    instance comStub: Svc.ComStub base id ComCcsdsConfig.BASE_ID + 0x0A000
 
     topology FramingSubtopology {
         # Usage Note:
@@ -131,6 +136,7 @@ module ComCcsds {
         instance framer
         instance spacePacketFramer
         instance apidManager
+        instance aggregator
 
         connections Downlink {
             # ComQueue <-> SpacePacketFramer
@@ -141,10 +147,15 @@ module ComCcsds {
             spacePacketFramer.bufferDeallocate -> commsBufferManager.bufferSendIn
             spacePacketFramer.getApidSeqCount  -> apidManager.getApidSeqCountIn
             # SpacePacketFramer <-> TmFramer
-            spacePacketFramer.dataOut -> framer.dataIn
+            spacePacketFramer.dataOut -> aggregator.dataIn
-            framer.dataReturnOut      -> spacePacketFramer.dataReturnIn
+            aggregator.dataOut        -> framer.dataIn
+
+            framer.dataReturnOut      -> aggregator.dataReturnIn
+            aggregator.dataReturnOut    -> spacePacketFramer.dataReturnIn
+
             # ComStatus
-            framer.comStatusOut            -> spacePacketFramer.comStatusIn
+            framer.comStatusOut            -> aggregator.comStatusIn
+            aggregator.comStatusOut        -> spacePacketFramer.comStatusIn
             spacePacketFramer.comStatusOut -> comQueue.comStatusIn
             # (Outgoing) Framer <-> ComInterface connections shall be established by the user
         }

Svc/Subtopologies/ComCcsds/ComCcsdsConfig/ComCcsdsConfig.fpp

@@ -4,19 +4,22 @@ module ComCcsdsConfig {
     
     module QueueSizes {
         constant comQueue    = 50
+        constant aggregator  = 10
     }
     
     module StackSizes {
         constant comQueue   = 64 * 1024
+        constant aggregator = 64 * 1024
     }
 
     module Priorities {
         constant comQueue   = 101
+        constant aggregator = 100
     }
 
     # Queue configuration constants
     module QueueDepths {
-        constant events      = 100             
+        constant events      = 200             
         constant tlm         = 500            
         constant file        = 100            
     }

ci/tests/fputil.bash

@@ -108,7 +108,7 @@ function integration_test_run {
                 TIMEOUT="gtimeout" # macOS homebrew "coreutils"
             fi
             TOP_CONFIG_ARGS="--deployment-config ${WORKDIR}/test/int/int_config.json"
-            ${TIMEOUT} --kill-after=10s 180s pytest ${DICTIONARY_ARGS} ${TOP_CONFIG_ARGS}
+            ${TIMEOUT} --kill-after=10s 300s pytest ${DICTIONARY_ARGS} ${TOP_CONFIG_ARGS}
         )
         RET_PYTEST=$?
         pkill -P $GDS_PID

default/config/ComCfg.fpp

@@ -17,6 +17,7 @@ module ComCfg {
     # - potentially APID enum ?
     constant SpacecraftId = 0x0044    # Spacecraft ID (10 bits)
     constant TmFrameFixedSize = 1024  # Needs to be at least COM_BUFFER_MAX_SIZE + (2 * SpacePacketHeaderSize) + 1
+    constant AggregationSize = TmFrameFixedSize - 6 - 6 - 1 - 2  # 2 header (6) + 1 idle byte + 2 trailer bytes
 
     @ APIDs are 11 bits in the Space Packet protocol, so we use U16. Max value 7FF
     enum Apid : FwPacketDescriptorType {