Detailed Description

The ServiceBlocks are written in code to build new event-based operators that need to use blocking calls like accessing databases, using sockets to send/receive data, accessing the file system, etc.

ServiceBlock allows processing message events in a background thread safely. For each input event ProcessAsync() function is called and desired output event can just be returned from the call to be sent.

There may be a need to do something either before the ProcessAsync() call, in the ServiceBlock's thread context or after the ProcessAsync() call, back in the ServiceBlock's thread context before emitting the output event.

• To add code before the ProcessAsync() is executed, override the ProcessEvent() function.
• To add code before an event is emitted after calling the Emit() function or when the ProcessAsync() function returns, override the ProcessEmit() function.

Properties

EventID and ArgumentID

There are two helper data structures: EventID and ArgumentID. These allow safe access to ServiceBlock data and pass event identifiers from/to asynchronous context. The EventID allows comparing with Event member in ProcessAsync() to identify the triggering event source when needed.

EventID WhatEvent::ProcessAsync(const EventID& eventInID, ArgumentData& data) const
{
  if (eventInID == GoodIn)
    return GoodOut;
  return BadOut;
}

The ArgumentID is used as a key in the ArgumentData data map. By default, it is populated by input Argument values so they can be safely accessed from the data map in ProcessAsync() function. The ArgumentID also has meta member that is for the end user to add any kind of meta-information about the data when adding custom data in ProcessEvent() function, like names of the Arguments etc. The ArgumentID also supports exposing data from other CDP value types via the ArgumentData data map.

bool CustomDataOp::ProcessEvent(const EventID& eventInID, ArgumentData& data)
{
  // Adding a new value of MyProperty and name as meta info to ArgumentData
  data[{{MyProperty}, {MyProperty.GetNodeName()}}] = MyProperty.GetVariantValue();

  // To add/update meta info for a value already in the map it needs to be extracted, updated and added back.
  // An alternative approach would be to clear the data map and repopulate it with desired items
  // for example when all arguments need names in meta info.
  auto item = data.extract(MyArgument);
  item.key().meta = MyArgument->GetNodeName();
  data.insert(std::move(item));

  return true;
}

EventID CustomDataOp::ProcessAsync(const EventID& eventInID, ArgumentData& data) const
{
  // Print out all items in data with meta info and value
  for (auto& mapPair : data)
    CDPMessage("Key.meta=%s, Value=%s\n", mapPair.first.meta.c_str(), mapPair.second.ToString().c_str());

  if ((int)data[MyProperty] > 0)
  {
    data[MyProperty] = 0;
    return GoodValue;
  }
  return BadValue;
}

bool CustomDataOp::ProcessEmit(const EventID& eventInID, const EventID& eventOutID, ArgumentData& data)
{
  // Update MyProperty member value from the data map
  MyProperty = (int)data[MyProperty];
  return true;
}

Example

The ReadFile ServiceBlock is intended to provide parallel and sequential support for reading lines from a given file. Upon the arrival of the Read event message, the block will start an asynchronous process to open and read line data to send with the Line event message. When reading is done the ServiceBlock emits a Done event message and when there is a file open error it emits an Error event message with error text.

The following is a list of library-includes, members and types that are used, with short comments about any configuration performed on them.

#include <iostream>
#include <fstream>
#include <string>

Sequencer::Event* Read;  // Input event with FileName argument selected in the event's Data member
Sequencer::Event* Line;  // Output event with FileName, TextLine argument selected in the event's Data member
Sequencer::Event* Done;  // Output event with FileName argument selected in the event's Data member
Sequencer::Event* Error; // Output event with FileName, ErrorText argument selected in the event's Data member

ArgumentBase* FileName;  // Input argument for file name
ArgumentBase* TextLine;  // Output argument to hold the line from the file for each Line event
ArgumentBase* ErrorText; // Output argument string to hold error message

The ProcessAsync() function is called from the worker thread context so accessing ServiceBlock members or containers directly is not advised. Argument values are made available for safe access via the ArgumentData data map. Event names can safely be used to emit events from the function, and Argument names can be used to access the Argument values in the data map.

void ReadFile::Configure(XMLPrimitive* operatorXML)
{
  ServiceBlock::Configure(operatorXML);
  FileName =  FindArgument("FileName");
  TextLine = FindArgument("TextLine");
  ErrorText = FindArgument("ErrorText");

  Read = FindEvent("Read");
  Line = FindEvent("Line");
  Done = FindEvent("Done");
  Error = FindEvent("Error");
}

EventID ReadFile::ProcessAsync(const EventID& eventInID, ArgumentData& data) const
{
  std::ifstream fileStream;
  std::string readLine;
  fileStream.open((std::string)data[FileName], std::ios::in);
  if (!fileStream)
  {
    data[ErrorText] = std::string(strerror(errno));
    return Error;
  }
  while (!fileStream.eof())
  {
    if (IsAbortRequested())
      return {}; // abort without output event
    std::getline(fileStream, readLine);
    if (readLine.empty() && fileStream.eof())
      return Done;
    data[TextLine] = readLine;
    Emit(Line, data);
  }
  return Done;
}