Tuesday, January 29, 2019

Pseudo Visual Compiler using Interface to Ada as the OFP - Part 4


Pseudo Visual Compiler using Interface to Ada as the OFP - Part 4

Following along with the translation of the C# message delivery framework to Ada of the previous posts, I am again reporting my progress.

For this report the use of the Delivery and Library packages will be described as well as showing the test components in a more realistic presentation.  A modification to Component to allow writes to the particular generic Disburse queue of a component will also be presented.  This will leave Remote, Receive, Transmit, and NamedPipe as the major classes get to be translated to Ada along with a portion of Format for remote messages.  These are the classes involved in the communication with remote applications and hence those necessary for the Ada application to communicate with the C# application and act as the extremely mini OFP to the C# CDU application for which I started this series of reports.

Delivery

Upon translating the C# Delivery class to Ada and then using it to deliver a message to a component, I ran into a problem.  In the previous test cases I had direct visibility to the other component's Disburse queues so the test code could just write a message to the particular component.

However, with the message delivery framework design, components do not have such direct visibility and the Delivery class/package has to write the message of a producing component to the consumer component of the message topic.  Then the Disburse EventWait is signaled to resume and the message is dequeued and treated by the component.  The problem was that the Delivery package couldn't determine which queue belonged to the consumer component nor did it have the necessary visibility to write the message to the queue since a feature of the design is that the components be self contained with only messages to interface with other components.

I therefore created a new interface to the Component package to look up the user component's queue.  This solved half the problem but it was still necessary to have visibility to the queue in order to Write to it.  That is, each of the queues were instantiated inside the particular component's package body to maintain the independence of one component from each of the others.

I first tried to move the instantiations of the queues inside of the Component package so it would have visibility to them.  That is, by creating a queue factory/farm.  However, this presented problems in Ada where pointers can exist to procedures and functions – the callbacks – but not to an instance of a package such as can be done in C# to the class instances.  For instance, passing the address of the queue to Component Register resulted in a value of 0 rather than the location of the instance of the generic package.  This made a queue farm problematic.  In addition, Ada instantiates the generics first so it would have done so for every queue in the farm whether the particular application needed the full set of queues.

I then thought how I could have a callback function in each component and register the location of the callback function when registering the component with Component.  Then, Delivery could invoke a new Component function to write the message.  Component could then look up the callback function associated with the consumer component (as passed by Delivery) and invoke it to pass the message to it.  Such a callback function could be located in each particular component and each such function would have visibility to the queue of the particular component.  Thus it could invoke the Write function of the instance of the queue.  Problem solved.

Another path would have been to get rid of the generic and have had the new Queues package under Consumer provide the data space for each component.  Of course, in the C# component classes, the queues are not declared as private.  Hence they are available for other components to write to them.  Hence Delivery has visibility to Write to them.  A no-no for the design concept, although ignored in the implementations, so something that needs to be corrected.

While I was thinking about message delivery, once again, it occurred to me that it should be pointed out that only the three and four parameter Publish methods should be used by a user component.  That is, the one and two parameter Publish methods are only to be used by the message delivery framework to route messages for/from remote delivery.  However, the two parameter and one parameter Publish methods are also public in C# since they must be invoked by other classes; namely Library, ReceiveInterface, and Transmit. 

It should be noted, except for needing access to EventWait, a periodic component that only produces messages only needs a Disburse queue to wakeup at the end of its interval.  One that wasn't periodic wouldn't need a queue at all – but such a component would be very unusual.

The matter of queuing the message via the callback raises the question of why not have the component treat the message then and there rather than writing it to the queue.  But this would treat it in the thread of the publishing component.  Higher priority publishers could suspend the processing of other messages causing inconsistent, partially modified component data.  A big no-no.

In would nice to surround the message delivery framework with a barrier to prevent access by user components.  This was done in the exploratory projects of years ago by placing them inside a framework umbrella.  In the interest of simplifying the framework, this was eliminated in the C# version.  It won't work for a user component to attempt to use these one and two parameter Publish methods since these Delivery methods don't attempt to look up user components to which to route the message.

While discussing the previous, much older exploratory project it could be pointed out, since it explored a wide variety of features, that it made allowances for very large messages that had to be divided into multiple transmissions by the framework.  This required keeping track of the partial messages to ensure that all had been delivered and reassembled in the correct order, etc.  The C# version had no such pretensions and was to be much smaller.  Therefore, if ever needed, it is left to the two components involved to implement the transmission and receipt of such messages.

As far as the three and four parameter methods are concerned, the three parameter method is just a short cut for every message except Response topics which must be delivered to the component that published the corresponding Request topic.  That is, the three parameter method immediately invokes the four parameter method with a null parameter for the component to which the message is to be delivered.  That is, Delivery uses Library to look up the component(s) to which the topic is to be delivered.  As described below, Library keeps track of all the components that are consumers of a topic via the registration of the topics by the components.  Therefore, a component could invoke the four parameter method by supplying a null value for the consumer component.

Delivery.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace VisualCompiler
{

    static public class Delivery
    {
        // This class implements a portion of the framework meant to deliver
        // messages (that is, instances of topics) to the components that
        // have registered to consume the topic.  This is straight forward
        // for the default topics with Publish looking up in the Library
        // those components that have registered to consume the topic.
        //
        // A Request/Response topic can have multiple components that publish
        // the request topic but only one consumer of the topic.  The consumer
        // component analyses the request and produces the response.  Delivery
        // must discover which component published the request and deliver the
        // response to that component.

        public enum Distribution
        {
            CONSUMER,
            PRODUCER
        };

        public struct HeaderType
        {
            public Int16 CRC;                     // message CRC
            public Topic.TopicIdType id;          // topic of the message
            public Component.ParticipantKey from; // publishing component
            public Component.ParticipantKey to;   // consumer component
            public Int64 referenceNumber;         // reference number of message
            public Int16 size;                    // size of data portion of message
        }

        public const Int16 HeaderSize = 16;

        // A message consists of the header data and the actual data of the
        // message
        public struct MessageType
        {
            public HeaderType header;
            public string data;
        }

        static public MessageType nullMessage;

        static public Int64 referenceNumber; // ever increasing message reference
                                             // number

        // Initialize data that would otherwise be done in a constructor.
        static public void Initialize()
        {
            referenceNumber = 0;

            nullMessage.Header.CRC = 0;
            nullMessage.header.from.appId = 0;
            nullMessage.header.from.comId = 0;
            nullMessage.header.from.subId = 0;
            nullMessage.header.to.appId = 0;
            nullMessage.header.to.comId = 0;
            nullMessage.header.to.subId = 0;
            nullMessage.header.referenceNumber = 0;
            nullMessage.header.size = 0;
            nullMessage.data = null;
        } // end Initialize

        static private void PublishResponseToRequestor
                            (Topic.TopicIdType topic,
                             Library.TopicTableType consumers,
                             MessageType msg)
        {
            bool found = false;
            for (int i = 0; i < consumers.count; i++)
            {
                if (Component.CompareParticipants(msg.header.to,
                                                  consumers.list[i].component))
                {
                    // Return response to the requestor
                    consumers.list[i].referenceNumber = 0;
                    Disburse queue =
                        Component.GetQueue(consumers.list[i].component);
                    if (queue != null)
                    {
                        queue.Write(msg);
                        found = true;
                        break; // exit inner loop
                    }
                }
            } // end for

            if (!found)
            {
                Console.WriteLine
                    ("ERROR: Delivery couldn't find requestor for response");
            }

        } // end PublishResponseToRequestor

        // Publish an instance of a topic message by a component
        static public void Publish(Topic.TopicIdType topic,
                                   Component.ParticipantKey component,
                                   string message)
        { // forward for treatment
            Publish(topic, component, Component.nullKey, message);
        } // Publish

        // Publish an instance of a response topic message by a component
        static public void Publish(Topic.TopicIdType topic,
                                   Component.ParticipantKey component,
                                   Component.ParticipantKey from,
                                   string message)
        {
            // Increment the reference number associated with all new messages
            referenceNumber++;

            // Initialize an instance of a message
            MessageType msg;
            msg.header.CRC = 0;
            msg.header.id = topic;
            msg.header.from = component;
            msg.header.to = from;
            msg.header.referenceNumber = referenceNumber;
            msg.header.size = (Int16)message.Length;
            msg.data = message;

            // Get the set of consumers of the topic
            Library.TopicTableType consumers = Library.TopicConsumers(topic);

            Topic.TopicIdType requestTopic = topic;
            if (topic.ext == Topic.Extender.RESPONSE) // the message has to be
                                                      //   delivered to the
            {                                         //   particular requestor
                // Get the consumer of the request topic
                requestTopic.ext = Topic.Extender.REQUEST;
                Library.TopicTableType requestConsumers =
                    Library.TopicConsumers(requestTopic);
                if (Component.CompareParticipants(msg.header.to,
                                                  Component.nullKey))
                {
                    Console.WriteLine("ERROR: No 'To' address for Response");
                    return;
                }
                if (msg.header.to.appId != App.applicationId)
                { // send to remote application
                    Publish(msg.header.to.appId, msg);
                    return;
                }

                PublishResponseToRequestor(topic, consumers, msg);

            } // end if published topic is a Response

            else if (topic.ext == Topic.Extender.REQUEST) // only one consumer
            {                                             //   possible
                if (consumers.count > 0)
                { // forward request to the lone consumer of request topic
                    msg.header.to = consumers.list[0].component;
                    consumers.list[0].requestor = component;
                    consumers.list[0].referenceNumber = referenceNumber;
                    if (msg.header.to.appId != App.applicationId)
                    { // send to remote app
                        Publish(msg.header.to.appId, msg);
                    }
                    else
                    { // forward to local consumer
                        bool found = false;
                        Disburse queue =
                            Component.GetQueue(consumers.list[0].component);
                        if (queue != null)
                        {
                            queue.Write(msg);
                            found = true;
                        }

                        if (!found)
                        {
                            Console.WriteLine(
                               "ERROR: Delivery didn't have queue for request");
                        }
                    }
                }
                else
                {
                    Console.WriteLine(
                        "ERROR: Delivery couldn't find consumer for request");
                }
            }

            else // the published topic has to be the Default -
            {    //   can be multiple consumers
                for (int i = 0; i < consumers.count; i++)
                {
                    msg.header.to = consumers.list[i].component;
                   
                    // Avoid sending topic back to the remote app that
                    // transmitted it to this app or forwarding a remote
                    // message that is to be delivered to a different
                    // component.
                    if (Ignore(msg, consumers.list[i].component))
                    {
                        // ignore
                    }
                    else // publish to local or remote component
                    {
                        if (msg.header.to.appId != App.applicationId)
                        { // Deliver message to remote application
                            Publish(msg.header.to.appId, msg);
                        }
                        else
                        { // Deliver message to local application by copying to
                          // consumer's queue
                            consumers.list[i].requestor = component;
                            consumers.list[i].referenceNumber = 0;
                            bool found = false;
                            Disburse queue =
                               Component.GetQueue(consumers.list[i].component);
                            if (queue != null)
                            {
                                queue.Write(msg);
                                found = true;
                            }
                            if (!found)
                            {
                                Console.WriteLine(
                                   "ERROR: local default Delivery couldn't find queue for consumer");
                            }
                        }
                    } // end if Ignore
                } // end for

            } // end if

        } // end Publish

        // Publish an instance of a remote topic message forwarded by Receive
        static public void Publish(MessageType message)
        {
            // Get the set of consumers of the topic
            Library.TopicTableType consumers =
                Library.TopicConsumers(message.header.id);

            if (message.header.id.ext == Topic.Extender.REQUEST)
            { // forward the request topic to its consumer
                for (int i = 0; i < consumers.count; i++)
                {
                    if (message.header.id.topic == consumers.list[i].id.topic)
                    { // the only possible consumer of the request topic 
                        consumers.list[i].requestor = message.header.from;
                        consumers.list[i].referenceNumber =
                            message.header.referenceNumber;
                        bool found = false;
                        Disburse queue =
                            Component.GetQueue(consumers.list[i].component);
                        if (queue != null)
                        {
                            queue.Write(message);
                            found = true;
                        }
                        if (!found)
                        {
                            Console.WriteLine(
                                "ERROR: remote Request Delivery couldn't find queue for consumer");
                        }
                        return; // can only be one consumer
                    }
                }
            } // end for
            else if (message.header.id.ext == Topic.Extender.RESPONSE)
            { // forward the response topic to the request publisher
                for (int i = 0; i < consumers.count; i++)
                {
                    Console.WriteLine("Delivery Response consumers {0} {1} {2} {3} {4} {5} {6}",
                        consumers.count, message.header.id.topic,
                        consumers.list[i].id.topic,
                        consumers.list[i].component.appId,
                        consumers.list[i].component.comId,
                        message.header.to.appId, message.header.to.comId);
                    if ((message.header.id.topic == consumers.list[i].id.topic)
                        && (Component.CompareParticipants(
                               consumers.list[i].component,
                               message.header.to)))
                    { // found the publisher of the Request 
                        bool found = false;
                        Disburse queue =
                            Component.GetQueue(message.header.to);
                        if (queue != null)
                        {
                            Console.WriteLine("queued the message");
                            queue.Write(message);
                            found = true;
                        }
                        if (!found)
                        {
                            Console.WriteLine(
                                "ERROR: Remote Response Delivery couldn't find queue for consumer");
                        }
                        break; // exit loop
                    }
                } // end for
            }
            else // Default topic - forward to possible multiple consumers
            {
                for (int i = 0; i < consumers.count; i++)
                {
                    if (message.header.id.topic == Topic.Id.HEARTBEAT)
                    {
                        Console.WriteLine("Deliver HEARTBEAT {0} {1} {2} {3}",
                            message.header.to.appId, message.header.to.comId,
                            message.header.from.appId,
                            message.header.from.comId);
                    }
                    // Avoid sending topic back to the remote app that
                    // transmitted it to this app or forwarding a remote
                    // message that is to be delivered to a different
                    // component.
                    if ((consumers.list[i].id.topic == message.header.id.topic)
                        &&
                        (Ignore(message.header.to, message.header.from,
                                consumers.list[i].component)))
                    {
                        Console.WriteLine("Remote message ignored {0} {1} {2} {3} {4} {5}",
                            message.header.to.appId, message.header.to.comId,
                            message.header.from.appId, message.header.from.comId,
                            consumers.list[i].component.appId,
                            consumers.list[i].component.comId);
                    }
                    else
                    { // Deliver message to local application by copying to its queue
                        consumers.list[i].requestor = message.header.from;
                        consumers.list[i].referenceNumber = 0;
                        if (consumers.list[i].component.appId ==
                            App.applicationId)
                        {
                            bool queueFound = false;
                            Disburse queue =
                               Component.GetQueue(consumers.list[i].component);
                            if (queue != null)
                            {
                                queue.Write(message);
                                queueFound = true;
                            }
                            if (!queueFound)
                            {
                                Console.WriteLine("ERROR: Remote default Delivery couldn't find queue for consumer");
                            }
                        }
                    } // end if Ignore
                }

            }

        } // end Publish (from remote)

        // Remote messages are to be ignored if the From and To components are
        // the same since this would transmit the message back to the remote app.
        // Remote messages are only to be forwarded to the To component and not
        // to all the components of the consumers list since separate messages
        // are sent by the remote component for each local consumer.
        static private bool Ignore(MessageType message,
                                   Component.ParticipantKey component)
        {
            bool equal = Component.CompareParticipants(
                                     message.header.from, message.header.to);
            if ((equal) && (message.header.to.appId != App.applicationId))
            { // same from and to component and remote message
                return true;
            }
            if (message.header.from.appId != App.applicationId)
            { // remote message; check if consumer component is 'to' participant
                if (!Component.CompareParticipants(message.header.to,
                                                   component))
                {
                    return true;
                }
            }
            return false;
        } // end Ignore
       
        // Remote messages are to be ignored if the From and To components are
        // the same since this would transmit the message back to the remote app.
        // Remote messages are only to be forwarded to the To component and not
        // to all the components of the consumers list since separate messages
        // are sent by the remote component for each local consumer.
        static private bool Ignore(Component.ParticipantKey to,
                                   Component.ParticipantKey from,
                                   Component.ParticipantKey component)
        {
            bool equal = Component.CompareParticipants(from, to);
            if ((equal) && (to.appId != App.applicationId))
            { // same from and to component and remote message
                return true;
            }
            if ((from.appId != App.applicationId) &&    // from is remote
                (component.appId == App.applicationId)) // component is local
            { // remote message; check if consumer component is 'to' participant
                if (!Component.CompareParticipants(to, component))
                {
                    return true;
                }
            }
            return false;
        } // end Ignore

        // Deliver message to remote app
        static public void Publish(int remoteAppId, MessageType message)
        {
            // Obtain instance of Transmit class to which the message is
            // to be delivered
            Transmit transmit = Remote.TransmitInstance(remoteAppId);
            Console.WriteLine("Publish to Transmit queue {0}",
                              transmit.queue.queueName);
            if (transmit == null)
            {
                Console.WriteLine(
                    "ERROR: No Transmit class instance for Publish");
            }

            else
            {
                // Increment the reference number associated with all new
                // messages
                referenceNumber++;
                message.header.referenceNumber = referenceNumber;

                // Get the queue associated with the instance of the class
                if (transmit.queue != null)
                {
                    Console.WriteLine("Publish to Remote app {0} {1} {2} {3}", // {4}",
                         message.header.id.topic, message.header.id.ext,
                         remoteAppId, //transmit.queue.queueTable.toAppId,
                         message.header.referenceNumber);
                    transmit.queue.Write(message);
                }
                else
                {
                    Console.WriteLine(
                        "ERROR: Transmit queue for remote transmit is null");
                }
            }
        } // end Publish

    } // end Delivery class

} // end namespace

Ada Delivery package

The visible interface:
with Component;
with Itf;
with Topic;

package Delivery is

  type DistributionType
  is ( CONSUMER,
       PRODUCER
     );

  procedure Initialize;
  -- Initialize Delivery package

  procedure Publish
  ( Message : in Itf.MessageType );
  -- Re-Publish message received from Remote

  procedure Publish
  ( RemoteAppId : in Itf.Int8;
    Message     : in out Itf.MessageType );
  -- Publish message to Remote such as Register Request

  procedure Publish
  ( TopicId      : in Topic.TopicIdType;
    ComponentKey : in Itf.ParticipantKeyType;
    Message      : in String );
  -- Publish local message except for Response message

  procedure Publish
  ( TopicId      : in Topic.TopicIdType;
    ComponentKey : in Itf.ParticipantKeyType;
    From         : in Itf.ParticipantKeyType;
    Message      : in String );
  -- Publish local Response message to specify the source of Request message

end Delivery;

The implementation:
with Disburse;
with Library;
with Remote;
with System;
with Text_IO;
with Unchecked_Conversion;

package body Delivery is

  package Int_IO is new Text_IO.Integer_IO( Integer ); -- for debug

  ReferenceNumber : Itf.Int32; -- ever increasing message reference number

  type QueueCallbackType
  -- Callback to execute a Queue Write
  is access function
  ( Message : in Itf.MessageType
  ) return Boolean;

  function to_Ptr is new Unchecked_Conversion
                         ( Source => System.Address,
                           Target => QueueCallbackType );

  procedure Initialize is
  begin -- Initialize

    ReferenceNumber := 0; -- ever increment for each message

  end Initialize;

  procedure PublishResponseToRequestor
  ( TopicId   : in Topic.TopicIdType;
    Consumers : in out Library.TopicTableType;
    Msg       : in Itf.MessageType
  ) is

    Found : Boolean := False;

    use type System.Address;

  begin -- PublishResponseToRequestor
    for I in 1..Consumers.Count loop
      if Component.CompareParticipants( Msg.Header.To,
                                        Consumers.List(I).ComponentKey )
      then -- return response to the requestor
        Consumers.List(I).ReferenceNumber := 0;
        Found := Component.DisburseWrite( Consumers.List(I).ComponentKey,
                                          Msg );
      end if;
    end loop;

    if not Found then
      Text_IO.Put_Line("ERROR: Delivery couldn't find requestor for response");
    end if;

  end PublishResponseToRequestor;

  -- Remote messages are to be ignored if the From and To components are
  -- the same since this would transmit the message back to the remote app.
  -- Remote messages are only to be forwarded to the To component and not
  -- to all the components of the consumers list since separate messages
  -- are sent by the remote component for each local consumer.
  function Ignore
  ( Message      : in Itf.MessageType;
    ComponentKey : in Itf.ParticipantKeyType
  ) return Boolean is

    Equal : Boolean;

    use type Itf.Int8;

  begin -- Ignore

    Equal := Component.CompareParticipants
             ( Message.Header.From, Message.Header.To);
    if Equal and then Message.Header.To.AppId /= Itf.ApplicationId.Id then
      -- same from and to component and remote message
      return True;
    end if;
    if Message.Header.From.AppId /= Itf.ApplicationId.Id then
      -- remote message; check if consumer component is 'to' participant
      if not Component.CompareParticipants
             ( Message.Header.To, ComponentKey )
      then
        return True;
      end if;
    end if;
    return False;

  end Ignore;

  -- Remote messages are to be ignored if the From and To components are
  -- the same since this would transmit the message back to the remote app.
  -- Remote messages are only to be forwarded to the To component and not
  -- to all the components of the consumers list since separate messages
  -- are sent by the remote component for each local consumer.
  function Ignore
  ( To           : in Itf.ParticipantKeyType;
    From         : in Itf.ParticipantKeyType;
    ComponentKey : in Itf.ParticipantKeyType
  ) return Boolean is

    Equal : Boolean;

    use type Itf.Int8;

  begin -- Ignore

    Equal := Component.CompareParticipants(From, To);
    if Equal and then To.AppId /= Itf.ApplicationId.Id then
      -- same from and to component and remote message
      return True;
    end if;
    if From.AppId /= Itf.ApplicationId.Id and then -- from is remote
       ComponentKey.AppId = Itf.ApplicationId.Id   -- component is local
    then -- remote message; check if consumer component is 'to' participant
      if not Component.CompareParticipants(To, ComponentKey) then
        return True;
      end if;
    end if;
    return False;

  end Ignore;

  procedure Publish
  ( Message : in Itf.MessageType
  ) is

    Consumers : Library.TopicTableType;
    Found     : Boolean;

    use type Itf.Int8;
    use type System.Address;
    use type Topic.Extender_Type;
    use type Topic.Id_Type;

  begin -- Publish

    -- Get the set of consumers of the topic
    Consumers := Library.TopicConsumers(Message.Header.Id);

    if Message.Header.Id.Ext = Topic.REQUEST then
      -- forward the request topic to its consumer
      for I in 1..Consumers.Count loop
        if Message.Header.Id.Topic = Consumers.List(I).Id.Topic then
          -- the only possible consumer of the request topic
          Consumers.List(I).Requestor := Message.Header.From;
          Consumers.List(I).ReferenceNumber := Message.Header.ReferenceNumber;
          Found := Component.DisburseWrite( Consumers.List(I).ComponentKey,
                                            Message );
          if not Found then
            Text_IO.Put
            ("ERROR: remote Request Delivery couldn't find queue for consumer ");
            Int_IO.Put(Integer(Message.Header.From.AppId));
            Text_IO.Put(" ");
            Int_IO.Put(Integer(Message.Header.From.ComId));
            Text_IO.Put(" ");
            Int_IO.Put(Integer(Message.Header.To.AppId));
            Text_IO.Put(" ");
            Int_IO.Put(Integer(Message.Header.To.ComId));
            Text_IO.Put(" Topic ");
            Int_IO.Put(Integer(Topic.Id_Type'pos(Message.Header.Id.Topic)));
            Text_IO.Put(" ");
            Int_IO.Put(Integer(Topic.Extender_Type'pos(Message.Header.Id.Ext)));
            Text_IO.Put_Line(" ");
          else
            return; -- can only be one consumer
          end if;
        end if;
      end loop;
    elsif Message.Header.Id.Ext = Topic.RESPONSE then
      -- forward the response topic to the request publisher
      for I in 1..Consumers.Count loop
        if Message.Header.Id.Topic = Consumers.List(I).Id.Topic and then
           Component.CompareParticipants(Consumers.List(I).ComponentKey,
                                         Message.Header.To)
        then -- found the publisher of the Request
          Found := Component.DisburseWrite( Consumers.List(I).ComponentKey,
                                            Message );
          if not Found then
            Text_IO.Put_Line(
              "ERROR: Remote Response Delivery couldn't find queue for consumer");
          end if;
          Exit; -- loop
        end if;
      end loop;
    else -- Default topic - forward to possible multiple consumers

      for I in 1..Consumers.Count loop

        if Message.Header.Id.Topic = Topic.HEARTBEAT then
          null;
        end if;
        -- Avoid sending topic back to the remote app that transmitted it to
        -- this app or forwarding a remote message that is to be delivered to
        -- a different component.
        if Consumers.List(I).Id.Topic = Message.Header.Id.Topic and then
           Ignore(Message.Header.To, Message.Header.From,
                  Consumers.List(I).ComponentKey)
        then
          null;
        else -- Deliver message to local application by copying to its queue
          Consumers.List(I).Requestor := Message.Header.From;
          Consumers.List(I).ReferenceNumber := 0;
          if Consumers.List(I).ComponentKey.AppId = Itf.ApplicationId.Id then
            Found := Component.DisburseWrite( Consumers.List(I).ComponentKey,
                                              Message );
            if not Found then
              Text_IO.Put_Line("ERROR: Remote default Delivery couldn't find queue for consumer");
              Int_IO.Put(Integer(Message.Header.From.AppId));
              Text_IO.Put(" ");
              Int_IO.Put(Integer(Message.Header.From.ComId));
              Text_IO.Put(" ");
              Int_IO.Put(Integer(Message.Header.To.AppId));
              Text_IO.Put(" ");
              Int_IO.Put(Integer(Message.Header.To.ComId));
              Text_IO.Put(" Topic ");
              Int_IO.Put(Integer(Topic.Id_Type'pos(Message.Header.Id.Topic)));
              Text_IO.Put(" ");
              Int_IO.Put(Integer(Topic.Extender_Type'pos(Message.Header.Id.Ext)));
              Text_IO.Put_Line(" ");
            end if;
          end if;
        end if; -- Ignore

      end loop;

    end if;

  end Publish; -- (from remote)

  procedure Publish
  ( RemoteAppId : in Itf.Int8;
    Message     : in out Itf.MessageType
  ) is

    use type System.Address;

  begin -- Publish

    null;
-->>> remove the above and add code when TransmitQueue available.

  end Publish;

  procedure Publish
  ( TopicId      : in Topic.TopicIdType;
    ComponentKey : in Itf.ParticipantKeyType;
    Message      : in String
  ) is
  begin -- Publish
    -- forward for treatment
    Publish(TopicId, ComponentKey, Component.NullKey, Message);
  end Publish;

  procedure Publish
  ( TopicId      : in Topic.TopicIdType;
    ComponentKey : in Itf.ParticipantKeyType;
    From         : in Itf.ParticipantKeyType;
    Message      : in String
  ) is

    Consumers : Library.TopicTableType;
    Found     : Boolean;
    Length    : Integer := 0;
    RequestConsumers : Library.TopicTableType;
    RequestTopic : Topic.TopicIdType;
    Msg : Itf.MessageType;

    use type Itf.Int8;
    use type Itf.Int16;
    use type System.Address;
    use type Topic.Extender_Type;

  begin -- Publish

    -- Increment the reference number associated with all new messages
    ReferenceNumber := referenceNumber + 1;

    -- Initialize an instance of a message
    Msg.Header.CRC := 0;
    Msg.Header.Id := TopicId;
    Msg.Header.From := ComponentKey;
    Msg.Header.To := From;
    Msg.Header.ReferenceNumber := ReferenceNumber;
    Found := False;
    for I in 1..Itf.Int16(Message'Length) loop
      Length := Integer(I);
      Msg.Data(Length) := Message(Length);
      Msg.Header.Size := I; -- in case there is no NUL
      if Message(Length) = ASCII.NUL then
        Found := True;
        Msg.Header.Size := I-1;
        exit; -- loop
      end if;
    end loop;
    if not Found then -- need to add trailing NUL in case message sent to C#
      Msg.Data(Length+1) := ASCII.NUL;
    end if;

    -- Get the set of consumers of the topic
    Consumers := Library.TopicConsumers(TopicId);

    RequestTopic := TopicId;
    if TopicId.Ext = Topic.RESPONSE then -- the message has to be delivered
                                         --   to the particular requestor
      -- Get the consumer of the request topic
      RequestTopic.Ext := Topic.REQUEST;
      RequestConsumers := Library.TopicConsumers(RequestTopic);
      if Component.CompareParticipants(Msg.Header.To, Component.NullKey) then
        Text_IO.Put_Line("ERROR: No 'To' address for Response");
        return;
      end if;
      if Msg.Header.To.AppId /= Itf.ApplicationId.Id then
        -- send to remote application
        Publish(Msg.Header.To.AppId, Msg);
        return;
      end if;

      PublishResponseToRequestor(TopicId, Consumers, Msg);

    elsif TopicId.Ext = Topic.REQUEST then -- only one consumer possible
      if Consumers.Count > 0 then
        -- forward request to the lone consumer of request topic
        Msg.Header.To := Consumers.List(1).ComponentKey;
        Consumers.List(1).Requestor := ComponentKey;
        Consumers.List(1).ReferenceNumber := ReferenceNumber;
        if Msg.Header.To.AppId /= Itf.ApplicationId.Id then
          -- send to remote app
          Publish(Msg.Header.To.AppId, Msg);
        else -- forward to local consumer
          Found := Component.DisburseWrite( Consumers.List(1).ComponentKey,
                                            Msg );
          if not Found then
            Text_IO.Put_Line("ERROR: Delivery didn't have queue for request");
          end if;
        end if;
      else
        Text_IO.Put_Line("ERROR: Delivery couldn't find consumer for request");
      end if;

    else -- the published topic has to be the Default - can be multiple consumers
      for I in 1..Consumers.Count loop
        Msg.Header.To := Consumers.List(I).ComponentKey;

        -- Avoid sending topic back to the remote app that transmitted it to
        -- this app or forwarding a remote message that is to be delivered to
        -- a different component.
        if Ignore(Msg, Consumers.List(I).ComponentKey) then
          null; -- ignore
        else -- publish to local or remote component
          if Msg.Header.To.AppId /= Itf.ApplicationId.Id then
            -- Deliver message to remote application
            Publish(Msg.Header.To.AppId, Msg);
          else -- Deliver message to local application by copying to
               -- consumer's queue
            Consumers.List(I).Requestor := ComponentKey;
            Consumers.List(I).ReferenceNumber := 0;
            Found := Component.DisburseWrite( Consumers.List(I).ComponentKey,
                                              Msg );
            if not Found then
              Text_IO.Put_Line(
                "ERROR: local default Delivery couldn't find queue for consumer");
            end if;
          end if;
        end if; -- Ignore
      end loop;

    end if;

  end Publish;

end Delivery;
Notice that one Publish for two parameters has been avoided for now awaiting the translation of the packages needed for communication with remote applications. 

The new interface to the Disburse queues of the components is illustrated by
          Found := Component.DisburseWrite( Consumers.List(1).ComponentKey,
                                            Msg );
where the C# version will need to be changed in a similar manner.  The changes to Component to pass in the callback to the Write function of the queue and to implement the DisburseWrite function are below.

  function Register
  ( Name       : in Itf.V_Medium_String_Type; -- name of component
    Period     : in Integer;      -- # of millisec at which Main() function to cycle
    Priority   : in Threads.ComponentThreadPriority; -- Requested priority of thread
    Callback   : in Topic.CallbackType; -- Callback() function of component
    Queue      : in System.Address; -- message queue of component
    QueueWrite : in System.Address  -- message queue Write function of component
  ) return RegisterResult is
where QueueWrite has been added.  Added to Register is
    ComponentTable.List(Location).QueueWrite := to_Write_Ptr(QueueWrite);
where to_Write_Ptr is
    function to_Write_Ptr is new Unchecked_Conversion
                                 ( Source => System.Address,
                                   Target => DisburseWriteCallback );
And where
  type DisburseWriteCallback
  -- Callback to execute the Write function of a participant component's
  -- Disburse queue
  is access function
  ( Message : in Itf.MessageType
    -- Message to be written to the queue
  ) return Boolean; -- indicates if Write was successful
is declared in the Component specification.

For DisburseWrite, the specification changed to have
    QueueWrite : DisburseWriteCallback;
    -- Callback to Write to the component's queue
added to the table data type and
  function DisburseWrite
  ( ComponentKey : in Itf.ParticipantKeyType;
    -- Component for message delivery
    Message      : in Itf.MessageType
    -- Message to be delivered
  ) return Boolean; -- true indicates successful write to queue
was added for Delivery to invoke.  The package body implements it as
  function DisburseWrite
  ( ComponentKey : in Itf.ParticipantKeyType;
    Message      : in Itf.MessageType
  ) return Boolean is
  begin -- DisburseWrite

    for I in 1..ComponentTable.Count loop
      if CompareParticipants(ComponentTable.List(I).Key, ComponentKey) and then
         ComponentTable.List(I).QueueWrite /= null
      then
        return ComponentTable.List(I).QueueWrite(Message);
      end if;
    end loop;
    return False; -- no component to which to write the message

  end DisburseWrite;

The above code of the Component package looks up the particular user component that registered its Disburse queue so that Deliver can write the message.  Upon finding the component it invokes its Disburse queue Write callback.  The Write function of the particular component then invokes the Write function of that Disburse queue's instantiation.  That is, the component has direct visibility to its specific Disburse queue so it can add the message to the queue.

Sample test/debug code for components will be illustrated later.  For now, the callbacks are each implemented as
    function DisburseWrite
    ( Message : in Itf.MessageType
    ) return Boolean is
    begin -- DisburseWrite
      return DisburseQueue.Write(Message => Message);
    end DisburseWrite;
where the DisburseQueue is privately declared in the component package's body.


Library

The Library class/package is meant to retain a table ("library") of message topics with the components that produce/publish them and those that consume them.  The full participant key of each component is retained which enables Delivery to establish whether any particular message is to be written to a local component's queue of the current application or whether it should (or also) be transmitted to a remote application of the configuration to be consumed by its component(s) by writing it to its(their) queue(s).  This is done via the RegisterTopic method.

It also contains RegisterRemoteTopics, RemoveRemoteTopics, SendRegisterRequest, and SendRegisterResponse to support the determination of the topics that a remote component can produce that can be forwarded for consumption by a local component.  That is, there is a REGISTER topic that is only to be used by the message delivery framework that is built (via SendRegisterRequest in conjunction with Format RegisterRequestTopic) and transmitted by SendRegisterRequest when a running remote application is discovered.  RegisterRemoteTopics is invoked from ReceiveInterface to add the topics to the Library topicTable and send the acknowledgement via SendRegisterResponse.  RemoveRemoteTopics is invoked from Receive and Transmit when there is a disconnect to allow the Register topic to be re-sent upon reconnection.  The remote application sends the register response back to the sending application upon receiving and processing the Register request via RegisterRemoteTopics where upon the Register message will no longer be sent.

RegisterTopic is invoked from the component Install procedures, including Remote, which are invoked at startup before component threads are assigned.  Therefore, like the Register procedure of Component, it cannot be suspended by a higher priority thread while updating the topicTable.

Other methods are Callbacks and TopicConsumers as well as IsValid that checks whether a component is registering (or ReceiveInterface or Transmit is referencing) a valid topic id/extension combination.  TopicConsumers is invoked by Delivery to determine to which components to write a message into their Disburse queue.  Callbacks is unused.

Library.cs

RegisterTopic needs to be corrected as in the new Ada package to correctly avoid allowing a duplicate consumer of a Request topic.  Note:  There is nothing currently to avoid a request topic consumer in multiple applications.  It is dependent upon the application architect to avoid such a circumstance.  If it does occur, the request will be treated by the local consumer.

RegisterTopic was also changed in the Ada version to avoid registering NONE and ANY, as well as REGISTER and HEARTBEAT and anything with an extension of FRAMEWORK.  However, this won't work completely since Remote.cs registers ANY, FRAMEWORK for the Transmit component.  Therefore, Library.cs needs a change to prevent user components for registering these topics and Library.adb will need a change to allow them when invoked from framework packages.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace VisualCompiler
{
    static public class Library
    {
        // A library of registered message topics with their producer
        // and consumer components.

        // Component data from registration as well as run-time status
        public struct TopicDataType
        {
            public Topic.TopicIdType id;               // complete topic identifier
            public Component.ParticipantKey component; // component that registers
                                                       //   the topic
            public Delivery.Distribution distribution; // whether consumed or produced
            public Callback fEntry;                    // callback, if any, to consume
                                                       //   the messages
            public Component.ParticipantKey requestor; // component that produced a
                                                       //   REQUEST topic
            public Int64 referenceNumber;              // reference number of a
                                                       //   REQUEST topic
        };

        public class TopicTableType
        {
            public int count; // Number of declared topics of the configuration
                              //   of applications
            public TopicDataType[] list = new
                TopicDataType[Configuration.MaxApplications*Component.MaxComponents];
            // will need to be expanded
        };

        // Library of topic producers and consumers
        static private TopicTableType topicTable = new TopicTableType();

        // Data of Remote Request topic
        public struct TopicListDataType
        {
            public Topic.TopicIdType topic;
            public Component.ParticipantKey component;
            //public Component.ParticipantKey requestor;
        };

        // List of topics
        public class TopicListTableType
        {
            public int count;
            public TopicListDataType[] list = new TopicListDataType[25];
        }

        public struct CallbackDataType
        {
            public Callback cEntry; // Any entry point to consume the message
        };

        public class CallbackTableType
        {
            public int count; // Number of messages in the table
            public CallbackDataType[] list = new
                                        CallbackDataType[Component.MaxComponents];
        }

        // Initialize. A replacement for a constructor.
        static public void Initialize()
        {
            topicTable.count = 0;
        } // end Initialize

        // Possible results of attempt to register a topic
        public enum AddStatus
        {
            SUCCESS,   // Topic added to the library
            DUPLICATE, // Topic already added for the component
            FAILURE,   // Topic not added
            NOTALLOWED // Topic not allowed, such as for second consumer of REQUEST
        };

        // Determine if supplied topic is a known pairing.
        static public bool ValidPairing(Topic.TopicIdType id)
        {
            for (int i = 0; i < Topic.TopicIds.count; i++)
            {
              if ((id.topic == Topic.TopicIds.list[i].topic) && // then known
                   (id.ext == Topic.TopicIds.list[i].ext))       //   topic pairing
              {
                    return true;
               }
            }
            return false;
        } // end ValidPairing

        // Add a topic with its component, whether producer or consumer, and
        // entry for consumer
        static public AddStatus RegisterTopic
                      (Topic.TopicIdType id, Component.ParticipantKey component,
                       Delivery.Distribution distribution, Callback fEntry)
        {
            // Determine if supplied topic is a known pairing.
            bool entryFound = false;
            entryFound = ValidPairing(id);
            if (!entryFound)
            {
                   return AddStatus.NOTALLOWED;
            }

            // Determine if topic has already been added to the library.
            entryFound = false;
            for (int i = 0; i < topicTable.count; i++)
            {
                if (id.topic == topicTable.list[i].id.topic) // topic id already in table
                { // Be sure this new registration isn't for a request consumer
                    if ((id.ext == topicTable.list[i].id.ext) &&
                        (id.ext == Topic.Extender.REQUEST) &&
                        (distribution == Delivery.Distribution.CONSUMER))
                    {
                        if (Component.CompareParticipants(component,
                            topicTable.list[i].component))
                        {
                            Console.WriteLine(
                                  "ERROR: Only one Consumer of a Request allowed {0} {1} {2}",
                                   topicTable.list[i].id.topic, component.appId,
                                   component.comId);
                            entryFound = true;
                            return AddStatus.NOTALLOWED;
                        }
                    }
                } // end if topic in table
            } // end for

            // Check that consumer component has a queue
            if (distribution == Delivery.Distribution.CONSUMER)
            {
                for (int k = 0; k < Component.componentTable.count; k++)
                {
                    if (Component.CompareParticipants(
                           component, Component.componentTable.list[k].key))
                    {
//                        if ((Component.componentTable.list[k].queue == null) &&
//                            (Component.componentTable.list[k].circularQueue == null) &&
//                            (Component.componentTable.list[k].disburseQueue == null))
                        if (Component.componentTable.list[k].queue == null)
                        {
                            return AddStatus.NOTALLOWED;
                        }
                    }
                } // end for
            } 

            if (!entryFound) // add the topic with its component to the table
            {
                int k = topicTable.count;
                topicTable.list[k].id = id;
                topicTable.list[k].component = component;
                topicTable.list[k].distribution = distribution;
                topicTable.list[k].fEntry = fEntry;
                topicTable.count++;
                return AddStatus.SUCCESS;
            }

            return AddStatus.FAILURE;

        } // end RegisterTopic function

        static public void RegisterRemoteTopics(int remoteAppId, Delivery.MessageType message)
        {
            // Check if topics from remote app have already been registered.
            Console.WriteLine("RegisterRemoteTopics {0} count {1}",
                remoteAppId, topicTable.count);
            for (int i = 0; i < topicTable.count; i++)
            {
 //               Console.WriteLine("Library Topics list {0} {1} {2} {3} {4}",
 //                   i, topicTable.list[i].component.appId,
 //                   topicTable.list[i].requestor.appId, topicTable.list[i].id.topic,
 //                   topicTable.list[i].id.ext);
                if (topicTable.list[i].component.appId == remoteAppId)
                {
                    Console.WriteLine("RegisterRemoteTopics already in table");
                    // Send Response to the remote app again.
                    SendRegisterResponse(remoteAppId);
                    return; // since topicTable already contains entries from remote app
                }

            }

            // Decode Register Request topic.
            Library.TopicListTableType topics = new Library.TopicListTableType();
            topics = Format.DecodeRegisterRequestTopic(message);
            // Console.WriteLine("Decode Register Request {0}", topics.count);

            // Add the topics from remote app as ones that it consumes.
            int index = topicTable.count;
            for (int i = 0; i < topics.count; i++)
            {
                // ignore local consumer being returned in Register Request
                if (topics.list[i].component.appId != App.applicationId)
                {
                    topicTable.list[index].id = topics.list[i].topic;
                    Console.WriteLine("RegisterRequest topic {0} {1} {2}",index,
                        topicTable.list[index].id.topic,
                        topicTable.list[index].id.ext);
                    topicTable.list[index].component.appId =
                        topics.list[i].component.appId;
                    topicTable.list[index].component.comId =
                        topics.list[i].component.comId;
                    topicTable.list[index].component.subId =
                        topics.list[i].component.subId;
                    topicTable.list[index].distribution =
                        Delivery.Distribution.CONSUMER;
                    topicTable.list[index].fEntry = null;
                    topicTable.list[index].requestor.appId = remoteAppId;
                    topicTable.list[index].requestor.comId = 0; // add for Request
                    topicTable.list[index].requestor.subId = 0; //  message sometime
                    topicTable.list[index].referenceNumber = 0;
                    index++;
                }
                else
                {
                    Console.WriteLine
                        ("ERROR: Register Request contains local component {0} {1}",
                         topics.list[i].component.appId,
                         topics.list[i].component.comId);
                }
            }
            topicTable.count = index;

            Console.WriteLine("topicTable after Decode");
            for (int i = 0; i < topicTable.count; i++)
            {
                Console.WriteLine("{0} {1} {2} {3} {4} {5}",
                    i, topicTable.list[i].id.topic,
                    topicTable.list[i].id.ext, topicTable.list[i].distribution,
                    topicTable.list[i].component.appId,
                    topicTable.list[i].component.comId );
            }

            // Send Response to the remote app.
            SendRegisterResponse(remoteAppId);

        } // end RegisterRemoteTopics

        static public void RemoveRemoteTopics(int remoteAppId)
        {
            Console.WriteLine("RemoveRemoteTopics {0} count {1}",
                remoteAppId, topicTable.count);
            int newCount = topicTable.count;
            int index = topicTable.count - 1;
            int newIndex;
            for (int i = 0; i < topicTable.count; i++)
            {
                if (topicTable.list[index].component.appId == remoteAppId)
                {
                    Console.WriteLine("RemoteTopic in table {0} {1}",
                        topicTable.list[index].id.topic,
                        topicTable.list[index].id.ext);
                    // Move up any entries that are after this one
                    newIndex = index;
                    for (int j = index + 1; j < newCount; j++)
                    {
                        topicTable.list[newIndex] = topicTable.list[j];
                        newIndex++;
                    }
                    newCount = newIndex;
                }
                index--;
            } // end for
            topicTable.count = newCount;

            Console.WriteLine("topicTable after Decode");
            for (int i = 0; i < topicTable.count; i++)
            {
                Console.WriteLine("{0} {1} {2} {3} {4} {5}",
                    i, topicTable.list[i].id.topic,
                    topicTable.list[i].id.ext, topicTable.list[i].distribution,
                    topicTable.list[i].component.appId,
                    topicTable.list[i].component.comId);
            }

        } // end RemoveRemoteTopics

        // Send the Register Request message to the remote app.  This
        // message is to contain the topics of the local app for which
        // there are consumers so that the remote app will forward
        // any of those topics that it publishes.
        static public void SendRegisterRequest(int remoteAppId)
        {
            // Build table of all non-framework topics that have local consumers.
            TopicTableType topicConsumers = new TopicTableType();
            for (int i = 0; i < topicTable.count; i++)
            {
                if ((topicTable.list[i].id.topic != Topic.Id.REGISTER) &&
                    (topicTable.list[i].id.ext != Topic.Extender.FRAMEWORK))
                {
                    if ((topicTable.list[i].distribution ==
                         Delivery.Distribution.CONSUMER) &&
                        (topicTable.list[i].component.appId == App.applicationId))
                    {
                        topicConsumers.list[topicConsumers.count] =
                            topicTable.list[i];
                        topicConsumers.count++;
                    }
                }
            }

            // Build Register Request topic of these topics.
            Delivery.MessageType message =
                Format.RegisterRequestTopic(remoteAppId, topicConsumers);

            Console.WriteLine("Publish of Register Request");
            Delivery.Publish(remoteAppId, message);
            // if this works then Format doesn't really need to fill in header.
            // or do a new Publish for this.

        } // end SendRegisterRequest

        static private void SendRegisterResponse(int remoteAppId)
        {
            Delivery.MessageType responseMessage;
            responseMessage.header.CRC = 0;
            responseMessage.header.id.topic = Topic.Id.REGISTER;
            responseMessage.header.id.ext = Topic.Extender.RESPONSE;
            responseMessage.header.from = Component.nullKey;
            responseMessage.header.from.appId = App.applicationId;
            responseMessage.header.to = Component.nullKey;
            responseMessage.header.to.appId = remoteAppId;
            responseMessage.header.referenceNumber = 0;
            responseMessage.header.size = 0;
            responseMessage.data = "";

            Delivery.Publish(remoteAppId, responseMessage);

        } // end SendRegisterResponse

        // Return list of callback consumers
        static public CallbackTableType Callbacks(Component.ParticipantKey id)
        {
            CallbackTableType EntryPoints = new CallbackTableType();
            EntryPoints.count = 0;
            for (int i = 0; i < topicTable.count; i++)
            {
                if ((Component.CompareParticipants(topicTable.list[i].component,
                                                   id)) &&
                    (topicTable.list[i].fEntry != null))
                {
                    EntryPoints.list[EntryPoints.count].cEntry =
                        topicTable.list[i].fEntry;
                    EntryPoints.count++;
                }
            }
            return EntryPoints;
        } // end Callbacks

        // Return list of consumers of the specified topic
        static public TopicTableType TopicConsumers(Topic.TopicIdType id)
        {
            //debug
            bool heartbeat = false;
            if (id.topic == Topic.Id.HEARTBEAT) heartbeat = true;
           
            TopicTableType topicConsumers = new TopicTableType();
            for (int i = 0; i < topicTable.count; i++)
            {
                if ((id.topic == topicTable.list[i].id.topic) &&
                    (id.ext == topicTable.list[i].id.ext))
                {
                    if (topicTable.list[i].distribution ==
                        Delivery.Distribution.CONSUMER)
                    {
                           if (heartbeat)
                           {
                               Console.Write("Consume Heartbeat {0} {1}",
                                   topicTable.list[i].component.appId,
                                   topicTable.list[i].component.comId);
                           }
                        topicConsumers.list[topicConsumers.count] =
                            topicTable.list[i];
                        topicConsumers.count++;
                    }
                }
            }
            return topicConsumers;
        }

    } // end Library class

} // end namespace

Ada Library package

The visible interface:
with Component;
with Configuration;
with Delivery;
with Itf;
with Topic;

package Library is

  -- A library of registered message topics with their producer
  -- and consumer components.

  -- Possible results of attempt to register a topic
  type AddStatus
  is ( SUCCESS,   -- Topic added to the library
       DUPLICATE, -- Topic already added for the component
       FAILURE,   -- Topic not added
       NOTALLOWED -- Topic not allowed, such as for second consumer of REQUEST
     );

  -- Component data from registration as well as run-time status
  type TopicDataType
  is record
    Id              : Topic.TopicIdType;
    -- complete topic identifier
    ComponentKey    : Itf.ParticipantKeyType;
    -- component that produces the topic
    Distribution    : Delivery.DistributionType;
    -- whether consumed or produced
    fEntry          : Topic.CallbackType;
    -- callback, if any to consume the messages
    Requestor       : Itf.ParticipantKeyType;
    -- component that produced REQUEST topic
    ReferenceNumber : Itf.Int32;
    -- reference number of a REQUEST topic
  end record;

  type TopicTableArrayType
  is array(1..Configuration.MaxApplications*Component.MaxComponents)
  of TopicDataType;

  type TopicTableType
  is record
    Count : Integer;
    -- Number of declared topics of the configuration of applications
    List  : TopicTableArrayType;
  end record;

  -- Data of Remote Request topic
  type TopicListDataType
  is record
    TopicId      : Topic.TopicIdType;
    ComponentKey : Itf.ParticipantKeyType;
  --Requestor    : Itf.ParticipantKeyType;
  end record;

  -- List of topics
  type TopicListTableArrayType
  is array(1..25)
  of TopicListDataType;

  type TopicListTableType
  is record
    Count : Integer;
    List  : TopicListTableArrayType;
  end record;

  procedure Initialize;

  function RegisterTopic
  ( Id           : in Topic.TopicIdType;
    ComponentKey : in Itf.ParticipantKeyType;
    Distribution : in Delivery.DistributionType;
    fEntry       : in Topic.CallbackType
  ) return AddStatus;

  function TopicConsumers
  ( Id : in Topic.TopicIdType
  ) return TopicTableType;

  -- Determine if supplied topic is a known pairing.
  function ValidPairing
  ( Id : in Topic.TopicIdType
  ) return Boolean;

end Library;

The implementation:
with Format;
with Itf;
with System;
with Text_IO;
with Topic;

package body Library is

  -- Library of topic producers and consumers
  TopicTable : TopicTableType;

  type CallbackDataType
  is record
    cEntry : Topic.CallbackType; -- Any entry point to consume the message
  end record;

  type CallbackDataArrayType
  is array(1..Component.MaxComponents) of CallbackDataType;

  type CallbackTableType
  is record
    Count : Integer;
    List  : CallbackDataArrayType;
  end record;

  procedure SendRegisterResponse
  (RemoteAppId : in Itf.Int8);

  procedure Initialize is
  begin -- Initialize
    TopicTable.Count := 0;
  end Initialize;

  -- Add a topic with its component, whether producer or consumer, and entry
  –- for consumer
  function RegisterTopic
  ( Id           : in Topic.TopicIdType;
    ComponentKey : in Itf.ParticipantKeyType;
    Distribution : in Delivery.DistributionType;
    fEntry       : in Topic.CallbackType
  ) return AddStatus is

    EntryFound
    : Boolean;

    use type Delivery.DistributionType;
    use type System.Address;
    use type Topic.Extender_Type;
    use type Topic.Id_Type;

  begin -- RegisterTopic

    -- Determine if supplied topic is a known pairing.
    EntryFound := ValidPairing(Id);
    if (not EntryFound) then
      return NOTALLOWED;
    end if;

    -- Determine if a framework topic.  That is, a user component shouldn't be
    -- registering these topics to the Library.
    if Id.Topic in Topic.NONE..Topic.REGISTER or else
       Id.Ext = Topic.FRAMEWORK
    then
      return NOTALLOWED;
    end if;

    -- Determine if topic has already been added to the library.
    -- Note: A REQUEST message of a particular Topic should only be registered
    --       for one consumer.  Delivery will only route the REQUEST to one
    --       consumer.  That is, REQUEST messages are paired with a RESPONSE
    --       message with only one component to be designated to produce the
    --       RESPONSE.  There can be multiple requestors and the response will
    --       be delivered to the requesting component.
    EntryFound := False;
    for I in 1..TopicTable.Count loop
      if Id.Topic = TopicTable.List(I).Id.Topic then -- topic id already in table
        -- Be sure this new registration isn't for a second request consumer
        if (Id.Ext = TopicTable.List(I).Id.Ext and then
            Id.Ext = Topic.REQUEST and then
            Distribution = Delivery.CONSUMER)
        then
          if not Component.CompareParticipants -- different components
             ( ComponentKey,
               TopicTable.List(I).ComponentKey ) and then
             TopicTable.List(I).Distribution = Delivery.CONSUMER -- 2nd Consumer
          then
            EntryFound := True;
            return NOTALLOWED;
          end if;
        end if;
      end if; -- topic in table
    end loop;

    -- Check that consumer component has a queue
    if Distribution = Delivery.CONSUMER then
      for K in 1..Component.ComponentTable.Count loop
        if Component.CompareParticipants(
             ComponentKey, Component.ComponentTable.List(K).Key)
        then
          if Component.ComponentTable.List(K).Queue = System.Null_Address then
            return NOTALLOWED;
          end if;
        end if;
      end loop;
    end if;

    if not EntryFound then -- add the topic with its component to the table
      declare
        K : Integer := TopicTable.Count + 1;
      begin
        TopicTable.List(K).Id := Id;
        TopicTable.List(K).ComponentKey := ComponentKey;
        TopicTable.List(K).Distribution := Distribution;
        TopicTable.List(K).fEntry := fEntry;
        TopicTable.Count := K;
        return SUCCESS;
      end;
    end if;

    return FAILURE;

  end RegisterTopic;

  procedure RegisterRemoteTopics
  ( RemoteAppId : in Itf.Int8;
    Message     : in Itf.MessageType
  ) is

    Index  : Integer;
    Topics : Library.TopicListTableType;

    use type Itf.Int8;

  begin -- RegisterRemoteTopics

    -- Check if topics from remote app have already been registered.
    for I in 1..TopicTable.Count loop
      if TopicTable.List(I).ComponentKey.AppId = RemoteAppId then
        Text_IO.Put_Line("RegisterRemoteTopics already in table");
        -- Send Response to the remote app again.
        SendRegisterResponse(RemoteAppId);
        return; -- since topicTable already contains entries from remote app
      end if;

    end loop;

    -- Decode Register Request topic.
    Topics := Format.DecodeRegisterRequestTopic(Message);

    -- Add the topics from remote app as ones that it consumes.
    Index := TopicTable.Count + 1;
    for I in 1..Topics.Count loop
      -- ignore local consumer being returned in Register Request
      if Topics.List(I).ComponentKey.AppId /= Itf.ApplicationId.Id then
        TopicTable.list(Index).Id := Topics.List(I).TopicId;
        TopicTable.List(Index).ComponentKey.AppId :=
          Topics.List(I).ComponentKey.AppId;
        TopicTable.List(Index).ComponentKey.ComId :=
          Topics.List(I).ComponentKey.ComId;
        TopicTable.List(Index).ComponentKey.SubId :=
          Topics.List(I).ComponentKey.SubId;
        TopicTable.List(Index).Distribution := Delivery.CONSUMER;
        TopicTable.List(Index).fEntry := null;
        TopicTable.List(Index).Requestor.AppId := Itf.Int8(RemoteAppId);
        TopicTable.List(Index).Requestor.ComId := 0; -- add for Request message
        TopicTable.List(Index).Requestor.SubId := 0; --  sometime
        TopicTable.List(Index).ReferenceNumber := 0;
        TopicTable.Count := Index;
      end if;
    end loop;

    -- Send Response to the remote app.
    SendRegisterResponse(RemoteAppId);

  end RegisterRemoteTopics;

  procedure RemoveRemoteTopics
  ( RemoteAppId : in integer
  ) is

    NewCount : Integer := TopicTable.Count;
    Index    : Integer := TopicTable.Count;
    NewIndex : Integer;

  begin -- RemoveRemoteTopics
    for I in 1..TopicTable.Count loop
      if (Integer(TopicTable.List(Index).ComponentKey.AppId) = RemoteAppId) then
        -- Move up any entries that are after this one
        NewIndex := Index;
        for J in Index+1..NewCount loop
          TopicTable.List(NewIndex) := TopicTable.List(j);
          NewIndex := NewIndex + 1;
        end loop;
        NewCount := NewIndex;
      end if;
      Index := Index - 1;
    end loop;
    TopicTable.Count := NewCount;

  end RemoveRemoteTopics;

  -- Send the Register Request message to the remote app.  This
  -- message is to contain the topics of the local app for which
  -- there are consumers so that the remote app will forward
  -- any of those topics that it publishes.
  procedure SendRegisterRequest
  ( RemoteAppId : in Itf.Int8
  ) is

    Message : Itf.MessageType;
    TopicConsumers : TopicTableType;

    use type Delivery.DistributionType;
    use type Itf.Int8;
    use type Topic.Extender_Type;
    use type Topic.Id_Type;

  begin -- SendRegisterRequest

    -- Build table of all non-framework topics that have local consumers.
    TopicConsumers.Count := 0;
    for I in 1..TopicTable.Count loop
      if TopicTable.List(I).Id.Topic /= Topic.REGISTER and then
         TopicTable.List(I).Id.Ext /= Topic.FRAMEWORK
      then
        if TopicTable.List(I).Distribution = Delivery.CONSUMER and then
           TopicTable.List(I).ComponentKey.AppId = Itf.ApplicationId.Id
        then
          TopicConsumers.Count := TopicConsumers.Count + 1;
          TopicConsumers.List(TopicConsumers.Count) := TopicTable.List(I);
        end if;
      end if;
    end loop;

    -- Build Register Request topic of these topics.
    Message := Format.RegisterRequestTopic(RemoteAppId, TopicConsumers);

    Text_IO.Put_Line("Publish of Register Request");
    Delivery.Publish(RemoteAppId, Message);
    -- if this works then Format doesn't really need to fill in header.
    -- or do a new Publish for this.

  end SendRegisterRequest;

  procedure SendRegisterResponse
  ( RemoteAppId : in Itf.Int8
  ) is

    ResponseMessage : Itf.MessageType;

  begin  -- SendRegisterResponse

    ResponseMessage.Header.CRC := 0;
    ResponseMessage.Header.Id.Topic := Topic.REGISTER;
    ResponseMessage.Header.Id.Ext := Topic.RESPONSE;
    ResponseMessage.Header.From := Component.nullKey;
    ResponseMessage.Header.From.AppId := Itf.ApplicationId.Id;
    ResponseMessage.Header.To := Component.nullKey;
    ResponseMessage.Header.To.AppId := RemoteAppId;
    ResponseMessage.Header.ReferenceNumber := 0;
    ResponseMessage.Header.Size := 0;
    ResponseMessage.Data(1) := ' ';

    Delivery.Publish( RemoteAppId, ResponseMessage );

  end SendRegisterResponse;

  -- Return list of callback consumers
  function Callbacks
  ( Id : in Itf.ParticipantKeyType
  ) return CallbackTableType is

    EntryPoints : CallbackTableType;

    use type Topic.CallbackType;

  begin -- Callbacks

    EntryPoints.Count := 0;
    for I in 1..TopicTable.Count loop
      if ((Component.CompareParticipants(TopicTable.List(I).ComponentKey, Id))
      and then
          (TopicTable.List(I).fEntry /= null))
      then
        EntryPoints.List(EntryPoints.Count).cEntry := TopicTable.List(I).fEntry;
        EntryPoints.Count := EntryPoints.Count + 1;
      end if;
    end loop;
    return EntryPoints;

  end Callbacks;

  -- Return list of consumers of the specified topic
  function TopicConsumers
  ( Id : in Topic.TopicIdType
  ) return TopicTableType is

    Heartbeat : Boolean := False;

    TopicConsumers : TopicTableType;

    use type Delivery.DistributionType;
    use type Topic.Extender_Type;
    use type Topic.Id_Type;

  begin -- TopicConsumers

    if (Id.Topic = Topic.HEARTBEAT) then
      Heartbeat := True;
    end if;

    TopicConsumers.Count := 0;
    for I in 1..TopicTable.Count loop
      if ((Id.Topic = TopicTable.List(I).Id.Topic) and then
          (Id.Ext = TopicTable.List(I).Id.Ext))
      then
        if (TopicTable.List(I).Distribution = Delivery.CONSUMER) then
          TopicConsumers.Count := TopicConsumers.Count + 1;
          TopicConsumers.List(TopicConsumers.Count) := TopicTable.List(I);
        end if;
      end if;
    end loop;

    return TopicConsumers;

  end TopicConsumers;

  function ValidPairing
  ( Id : in Topic.TopicIdType
  ) return Boolean is

    use type Topic.Id_Type;
    use type Topic.Extender_Type;

  begin -- ValidPairing

    for I in 1..Topic.TopicIds.Count loop
      if ((Id.Topic = Topic.TopicIds.List(I).Topic) and then -- then known
          (Id.Ext = Topic.TopicIds.List(I).Ext)) then        --   topic pairing
        return True;
      end if;
    end loop;
    return False;

  end ValidPairing;

end Library;

Test/Debug samples of user components

Since the user components have been changed to have the Disburse queues be private to the component with Delivery necessary to forward messages to the component, a sample of these more realistic components will be provided.  One is Component4 that sends a Request message to Component5 (although it isn't necessary for it to know that Component5 will be the recipient) and Component5 to illustrate publishing the Response.

The following is the test procedure that specifies that this application is for the second application of the configuration (in anticipation that the C# application will be the first application); needed initializes of the Topic and Component packages; and the invocation of the Launch procedures of 5 user components.  Finally there is the creation of the Threads which comes last since there isn't any return and also because it is desirable not to have multiple threads running while doing the setup.  This will all be updated when finished to show the setup of a complete application.
  procedure Test1 is

    function to_Callback is new Unchecked_Conversion
                                ( Source => System.Address,
                                  Target => Topic.CallbackType );
  begin -- Test1

    Itf.ApplicationId.Id := 2;
    Topic.Initialize;
    -- Register the test components
    Component.Initialize;
    Component1a.Launch;
    Component2a.Launch;
    Component3a.Launch;
    Component4a.Launch;
    Component5a.Launch;

    Threads.Create; -- Do this after the threads have been created.

  end Test1;

Two component specs along with a conversion function for output of text strings and the component bodies follow.
  -- fourth Component thread for test1
  package Component4a is

    procedure Launch;

  end Component4a;

  -- fifth Component thread for test1
  package Component5a is

    procedure Launch;

  end Component5a;

  function to_Topic( Id : Topic.TopicIdType ) return String is
    Temp : String(1..19) := (others => ' ');
  begin
    case Id.Topic is
      when Topic.NONE  => Temp(1..4) := "NONE";
      when Topic.ANY   => Temp(1..3) := "ANY";
      when Topic.HEARTBEAT => Temp(1..9) := "HEARTBEAT";
      when Topic.REGISTER  => Temp(1..8) := "REGISTER";
      when Topic.TEST  => Temp(1..4) := "TEST";
      when Topic.TEST2 => Temp(1..5) := "TEST2";
      when Topic.TRIAL => Temp(1..5) := "TRIAL";
      when Topic.DATABASE => Temp(1..8) := "DATABASE";
      when Topic.OFP   => Temp(1..3) := "OFP";
    end case;
    case Id.Ext is
      when Topic.FRAMEWORK => Temp(11..19) := "FRAMEWORK";
      when Topic.DEFAULT   => Temp(11..17) := "DEFAULT";
      when Topic.TABLE     => Temp(11..15) := "TABLE";
      when Topic.KEYPUSH   => Temp(11..17) := "KEYPUSH";
      when Topic.REQUEST   => Temp(11..17) := "REQUEST";
      when Topic.RESPONSE  => Temp(11..18) := "RESPONSE";
    end case;
    return Temp;
  end to_Topic;

  . . .

These user components (where "user" is used as a qualifier to distinguish them from message delivery framework components such as Transmit) each have a queue name and a component name, a location to save the component's key as returned from Register, topic objects for each of the TopicRegister calls, the declarations of the AnyMessage message callback, the DisburseWrite callback, and the forever loop Main procedure callback necessary in Ada to be declared before referenced, the instantiation of the generic Disburse package, the Launch procedure (the only public item as in the package spec), and the code for the three callbacks.

The Launch procedure does the Register of the user component with Component which returns the Result.  Of this, if the status is Valid indicating success, the returned component key is saved, the wait event is provided to the queue, and the topics that the component will produce and wants to consume are registered with the Library.  After that Threads will run the component in its Main callback in its particular thread.  This callback will then await the end-of-wait signal that can either be when the periodic timer interval is satisfied if it is a periodic component or, otherwise, a new message has been written to the Disburse queue.  With messages now being delivered to message callbacks, this Main callback can be used to take other actions as necessary for the component.

Note that for Component4a (or any other periodic component) the messages will be treated in the message callback without regard to when the Main procedure callback is invoked.  That is, a higher frequency message topic(s) could be received more than once before the Main procedure is executed.  If the Main callback wants to treat each instance of the message, the message callback will need to put them into a table.  For a non-periodic such as Component5a, the Main loop should get invoked each time a message is received.  That is, Event Wait should be signaled causing the Main loop to execute.

  -- fourth Component thread for test1
  package body Component4a is

    Queue4 : Itf.V_Short_String_Type
           := ( Count => 2,
                Data  => "Q4                  " );

    Key : Itf.ParticipantKeyType := Component.NullKey;
    -- Component's key returned from Register

    RequestTopic1 : Topic.TopicIdType;
    RequestTopic2 : Topic.TopicIdType;
    RequestTopic3 : Topic.TopicIdType;

    procedure AnyMessage
    ( Message : in Itf.MessageType );

    package DisburseQueue
    -- Instantiate disburse queue for component
    is new Disburse( QueueName => Queue4'Address,
                     Periodic  => True,
                     Universal => AnyMessage'Address,
                     Forward   => System.Null_Address );

    function DisburseWrite
    -- Callback to write message to the DisburseQueue
    ( Message : in Itf.MessageType
    ) return Boolean;

    ComName4 : Itf.V_Medium_String_Type
    := ( Count => 4,
         Data  => "Com4                                              " );

    Result : Component.RegisterResult;

    OutIteration : Integer := 0;

    procedure Main -- callback
    ( Topic : in Boolean := False
    );


    procedure Launch is

      Status : Library.AddStatus;

      use type Component.ComponentStatus;
      use type Library.AddStatus;

      function to_Callback is new Unchecked_Conversion
                                  ( Source => System.Address,
                                    Target => Topic.CallbackType );

    begin -- Launch

      Result :=
        Component.Register
        ( Name       => ComName4,
          Period     => 4000, -- 4 sec period
          Priority   => Threads.LOWER, -- Requested priority of thread
          Callback   => to_Callback(Main'Address), -- Callback of component
          Queue      => DisburseQueue.Location,
          QueueWrite => DisburseWrite'Address );
      if Result.Status = Component.VALID then
        DisburseQueue.ProvideWaitEvent( Event => Result.Event );
        Key := Result.Key;
        RequestTopic1.Topic := Topic.DATABASE;
        RequestTopic1.Ext := Topic.REQUEST;
        Status := Library.RegisterTopic( RequestTopic1, Result.Key,
                                         Delivery.PRODUCER,
                                         to_Callback(Main'Address) );
        if Status /= Library.SUCCESS then
          Text_IO.Put_Line( "ERROR: Register of first Topic failed" );
        end if;
        RequestTopic2.Topic := Topic.OFP;
        RequestTopic2.Ext := Topic.KEYPUSH;
        Status := Library.RegisterTopic( RequestTopic2, Result.Key,
                                         Delivery.PRODUCER,
                                         to_Callback(Main'Address) );
        if Status /= Library.SUCCESS then
          Text_IO.Put_Line( "ERROR: Register of second Topic failed" );
        end if;
        RequestTopic3.Topic := Topic.DATABASE;
        RequestTopic3.Ext := Topic.RESPONSE;
        Status := Library.RegisterTopic( RequestTopic3, Result.Key,
                                         Delivery.CONSUMER,
                                         to_Callback(Main'Address) );
        if Status /= Library.SUCCESS then
          Text_IO.Put_Line( "ERROR: Register of third Topic failed" );
        end if;
      end if;

    end Launch;

    procedure DeliverToCom5 is

      OutMessage : Itf.MessageType;

    begin -- DeliverToCom5

      -- Write 1st message to Component5
      Text_IO.Put_Line("Com4 sending DATABASE message to Com5 ");
      OutMessage.Data(1..4) := "C4 D";
      OutMessage.Data(5) := ASCII.NUL;
      Delivery.Publish( TopicId      => ( Topic.DATABASE, Topic.REQUEST ),
                        ComponentKey => Key,
                        Message      => OutMessage.Data );

      -- Write 2nd message to Component5
      Text_IO.Put_Line("Com4 sending KEYPUSH message to Com5 ");
      OutMessage.Data(1..4) := "C4 K";
      OutMessage.Data(5) := ASCII.NUL;
      Delivery.Publish( TopicId      => ( Topic.OFP, Topic.KEYPUSH ),
                        ComponentKey => Key,
                        Message      => OutMessage.Data );

    end DeliverToCom5;

    procedure Main -- component callback
    ( Topic : in Boolean := False
    ) is

      Success : Boolean;

      Timer_Sec
      -- System time seconds as ASCII
      : String(1..2);
      System_Time
      -- System time
      : ExecItf.System_Time_Type;

    begin -- Main

      Text_IO.Put_Line("in Component4a callback");
      loop -- wait for event

        DisburseQueue.EventWait;

        System_Time := ExecItf.SystemTime;
        CStrings.IntegerToString(System_Time.Second, 2, False, Timer_Sec, Success);
        Text_IO.Put("C4 ");
        Text_IO.Put_Line(Timer_Sec(1..2));

        -- Write messages to Component5
        DeliverToCom5;

      end loop;

    end Main;

    function DisburseWrite
    ( Message : in Itf.MessageType
    ) return Boolean is
    begin -- DisburseWrite
      return DisburseQueue.Write(Message => Message);
    end DisburseWrite;

    -- Treat any message
    procedure AnyMessage
    ( Message : in Itf.MessageType
    ) is

      Success : Boolean;
      Iteration : String(1..4);

    begin -- AnyMessage

      Text_IO.Put("Entered Component4a AnyMessage ");

      CStrings.IntegerToString(Message.Header.ReferenceNumber, 4, False,
                               Iteration, Success);
      Int_IO.Put( Integer(Topic.Id_Type'pos(Message.Header.Id.Topic)) );
      Text_IO.Put(" ");
      Text_IO.Put( to_Topic(Message.Header.Id) );
      Text_IO.Put(Iteration(1..4));
      Text_IO.Put(" ");
      Text_IO.Put_Line(Message.Data(1..Integer(Message.Header.Size)));

    end AnyMessage;

  end Component4a;  -- fourth Component thread for test1

This fifth component also creates a message topic ForwardTable to specify individual callbacks when each of three topics are received.  The Disburse queue will invoke the callback of the topic when the topic is written and wait event is signaled to terminate the wait.  When Topic2, the Request message is received, the Topic2 message callback, publishes the Response message to be returned to the From component.  Although it can check what ComId that component has, the fifth component knows no more about it. 

Note however how Publish of the Response to the Request has to supply the From component.  Whereas the other Publish calls can use the three parameter Publish.  The publish of the response has to supply the From so that Delivery can determine which of multiple possible components sent the request so it can return the response to that particular requestor.

  -- fifth Component thread for test1
  package body Component5a is

    --<< visible so Component2 can write messages to the queue >>>
    Queue5 : Itf.V_Short_String_Type
           := ( Count => 2,
                Data  => "Q5                  " );

    Key : Itf.ParticipantKeyType := Component.NullKey;
    -- Component's key returned from Register

    RequestTopic1 : Topic.TopicIdType;
    RequestTopic2 : Topic.TopicIdType;
    RequestTopic3 : Topic.TopicIdType;
    RequestTopic4 : Topic.TopicIdType;

    procedure AnyMessage
    ( Message : in Itf.MessageType );

    -- Declared here so can be referenced to send messages to it
    ForwardTable : Itf.DisburseTableType;

    package DisburseQueue
    -- Instantiate disburse queue for component
    is new Disburse( QueueName => Queue5'Address,
                     Periodic  => False,
                     Universal => AnyMessage'Address,
                     Forward   => ForwardTable'Address );

    function DisburseWrite
    -- Callback to write message to the DisburseQueue
    ( Message : in Itf.MessageType
    ) return Boolean;

    procedure Main -- callback
    ( Topic : in Boolean := False
    );

    ComName5 : Itf.V_Medium_String_Type
    := ( Count => 4,
         Data  => "Com5                                              " );

    Result : Component.RegisterResult;

    procedure Topic1( Message : in Itf.MessageType );
    procedure Topic2( Message : in Itf.MessageType );
    procedure Topic3( Message : in Itf.MessageType );


    procedure Launch is

      Status : Library.AddStatus;

      use type Component.ComponentStatus;
      use type Library.AddStatus;

      function to_Callback is new Unchecked_Conversion
                                  ( Source => System.Address,
                                    Target => Topic.CallbackType );
      function to_Forward is new Unchecked_Conversion
                                 ( Source => System.Address,
                                   Target => Itf.ForwardType );
    begin -- Launch

      ForwardTable.Count := 3;
      ForwardTable.List(1).TopicId := ( Topic.TEST2, Topic.DEFAULT );
      ForwardTable.List(1).Forward := to_Forward(Topic1'Address);
      ForwardTable.List(2).TopicId := ( Topic.DATABASE, Topic.REQUEST );
      ForwardTable.List(2).Forward := to_Forward(Topic2'Address);
      ForwardTable.List(3).TopicId := ( Topic.OFP, Topic.KEYPUSH );
      ForwardTable.List(3).Forward := to_Forward(Topic3'Address);

      Result :=
        Component.Register
        ( Name       => ComName5,
          Period     => 0, -- not periodic
          Priority   => Threads.LOWER, -- Requested priority of thread
          Callback   => to_Callback(Main'Address), -- Callback() function of component
          Queue      => DisburseQueue.Location,
          QueueWrite => DisburseWrite'Address );
      if Result.Status = Component.VALID then
        DisburseQueue.ProvideWaitEvent( Event => Result.Event );
        Key := Result.Key;
        RequestTopic1.Topic := Topic.TEST2;
        RequestTopic1.Ext := Topic.DEFAULT;
        Status := Library.RegisterTopic( RequestTopic1, Result.Key,
                                         Delivery.CONSUMER,
                                         to_Callback(Main'Address) );
        if Status /= Library.SUCCESS then
          Text_IO.Put_Line( "ERROR: Register of first Topic failed" );
        end if;
        RequestTopic2.Topic := Topic.DATABASE;
        RequestTopic2.Ext := Topic.REQUEST;
        Status := Library.RegisterTopic( RequestTopic2, Result.Key,
                                         Delivery.CONSUMER,
                                         to_Callback(Main'Address) );
        if Status /= Library.SUCCESS then
          Text_IO.Put_Line( "ERROR: Register of second Topic failed" );
        end if;
        RequestTopic3.Topic := Topic.OFP;
        RequestTopic3.Ext := Topic.KEYPUSH;
        Status := Library.RegisterTopic( RequestTopic3, Result.Key,
                                         Delivery.CONSUMER,
                                         to_Callback(Main'Address) );
        if Status /= Library.SUCCESS then
          Text_IO.Put_Line( "ERROR: Register of third Topic failed" );
        end if;
        RequestTopic4.Topic := Topic.DATABASE;
        RequestTopic4.Ext := Topic.RESPONSE;
        Status := Library.RegisterTopic( RequestTopic4, Result.Key,
                                         Delivery.PRODUCER,
                                         to_Callback(Main'Address) );
        if Status /= Library.SUCCESS then
          Text_IO.Put_Line( "ERROR: Register of fourth Topic failed" );
        end if;
     end if;

    end Launch;

    function DisburseWrite
    ( Message : in Itf.MessageType
    ) return Boolean is
    begin -- DisburseWrite
      return DisburseQueue.Write(Message => Message);
    end DisburseWrite;

    procedure Main -- callback
    ( Topic : in Boolean := False
    ) is

      Success : Boolean;

      Timer_Sec
      -- System time seconds as ASCII
      : String(1..2);
      System_Time
      -- System time
      : ExecItf.System_Time_Type;

    begin -- Main
      Text_IO.Put_Line("in Component5a callback");
      loop -- wait for event
        DisburseQueue.EventWait;

        Text_IO.Put_Line("Com5a after end of wait ");

        System_Time := ExecItf.SystemTime;
        CStrings.IntegerToString(System_Time.Second, 2, False, Timer_Sec, Success);
        Text_IO.Put("C5 ");
        Text_IO.Put_Line(Timer_Sec(1..2));

      end loop;
    end Main;

    -- Treat any message of the component that doesn't have its own procedure
    procedure AnyMessage
    ( Message : in Itf.MessageType
    ) is

      Success : Boolean;
      Iteration : String(1..4);

    begin -- AnyMessage
      Text_IO.Put("Entered Component5 AnyMessage ");
      CStrings.IntegerToString(Message.Header.ReferenceNumber, 4, False,
                               Iteration, Success);
      Text_IO.Put( to_Topic(Message.Header.Id) );
      Text_IO.Put(Iteration(1..4));
      Text_IO.Put(" ");
      Text_IO.Put_Line(Message.Data(1..2));
    end AnyMessage;

    procedure Topic1( Message : in Itf.MessageType ) is
    begin -- Topic1

      Text_IO.Put( "Entered Component5 Topic1 ");
      Int_IO.Put( Integer(Topic.Id_Type'pos(Message.Header.Id.Topic)) );
      Text_IO.Put(" ");
      Text_IO.Put( to_Topic(Message.Header.Id) );
      Text_IO.Put_Line( Message.Data(1..2) );

     end Topic1;

    procedure Topic2( Message : in Itf.MessageType ) is
    begin -- Topic2

      Text_IO.Put( "Entered Component5 Topic2 ");
      Int_IO.Put( Integer(Topic.Id_Type'pos(Message.Header.Id.Topic)) );
      Text_IO.Put(" ");
      Text_IO.Put_Line( Message.Data(1..2) );

      Text_IO.Put( "Publish RESPONSE to be returned to Request publisher ");
      Int_IO.Put( Integer(Topic.Id_Type'pos(Message.Header.Id.Topic)) );
      Text_IO.Put(" ");
      Text_IO.Put( to_Topic(Message.Header.Id) );
      Text_IO.Put_Line( "C5 D" );
      Delivery.Publish
      ( TopicId      => ( Topic.DATABASE, Topic.RESPONSE ),
        ComponentKey => Key,
        From         => Message.Header.From, -- requesting component
        Message      => "C5 R" );

    end Topic2;

    procedure Topic3( Message : in Itf.MessageType ) is
    begin -- Topic3

      Text_IO.Put( "Entered Component5 Topic3 ");
      Int_IO.Put( Integer(Topic.Id_Type'pos(Message.Header.Id.Topic)) );
      Text_IO.Put(" ");
      Text_IO.Put( to_Topic(Message.Header.Id) );
      Text_IO.Put_Line( Message.Data(1..2) );

    end Topic3;

  end Component5a; -- fifth Component thread for test1

Sample Text Output

In this sample, Component4a sends its Database Request message for which Component5a happens to be the consumer (1a).  Then at 1b, Component5a has received the request in its Topic2 message callback and is going to publish the Database Response.  At "2 response returned" Component4a has received the response.
Com2 queued message to Com1 and 3
Deliver message           6 C2
C4 42
Com4 sending DATABASE message to Com5          ß 1a
Disburse Write Q1
Deliver message           7 C4
Deliver message           4 C2
Disburse Write Q3
doing EventWait Q3         168
Disburse Write Q5
doing EventWait Q5         176
Entered ForwardMessage
Invoking Universal for C2
Entered Component3a AnyMessage Entered ForwardMessage
Found Topic and Ext match in Forward Table
Entered Component5 Topic2           7 C4
Publish RESPONSE to be returned to Request publisher           7 DATABASE  REQUEST  C5 D ß 1b
Com4 sending KEYPUSH message to Com5           ß a
          4 TEST      DEFAULT   Deliver message           7 C5 ß
Disburse Write Q4
   6 C2
Deliver message           8 Com5a after end of wait
C3 42
C5 42
Com3a sending TEST message to Com5           1 ß b
C4
Deliver message           5 C3
Disburse Write Q5
doing EventWait Q5         176
Disburse Write Q5
doing EventWait Q5         176
Entered ForwardMessage
Found Topic and Ext match in Forward Table
Entered Component5 Topic3           8 OFP       KEYPUSH  C4  ß a rec'd
Entered ForwardMessage
Found Topic and Ext match in Forward Table
Entered Component5 Topic1           5 TEST2     DEFAULT  C3  ß b rec'd
Com5a after end of wait
C5 42
Com5a after end of wait
C5 42
Entered ForwardMessage
Invoking Universal for C2
Entered Component1a AnyMessage           6 TRIAL     REQUEST      4 C2 ß Com1 rec'd message
C1 42
Entered ForwardMessage
Invoking Universal for C5
Entered Component4a AnyMessage           7 DATABASE  RESPONSE    7 C5 R  ß 2 response returned
C4 42
Com4 sending DATABASE message to Com5


Posted by at
Labels: , ,

No comments:

Post a Comment

Subscribe to: Post Comments (Atom)