Securely extending data processing pipeline functionality

A computer program product for use at a computer system, the computer system including a data pipeline comprising a sequential arrangement of a plurality of pipeline modules, an electronic message received at the data pipeline being sequential processed by pipeline modules such that the output of at least one pipeline module is the input to the next sequential pipeline module, the computer program product for implementing a method for invoking an agent runtime to customize the functionality of a data pipeline, the computer program product comprising computer-readable storage medium having stored thereon computer-executable instructions that, when executed by a processor, cause the computer system to perform the method, including the following:

• receive electronic message related data at a pipeline module in the sequential arrangement of pipeline modules;
• detect the occurrence of an event at the pipeline module in response to processing the electronic message related data at the pipeline module;
• mapping the detected event to a specified topic string representing an electronic messaging command;
• scan configuration data for stored agents to attempt to identify any stored agents having topic strings corresponding to the specified topic string;
• indentify an agent having a topic string corresponding to the specified topic string, the agent having the corresponding topic string indicating that the agent is self-configured to hook into the detected event;
• asynchronously invoke the identified agent such that the identified agent can be executed to manipulate data related to the detected event in accordance with the configuration data for the identified agent, asynchronously invoking the identified agent including supplying appropriate values to an invoke API for the identified agent, supplying appropriate values including supplying values for;
• an execution context field for storing an execution context value that identifies an execution context object for an agent runtime;
• an event topic field for storing an event topic value that represents a type of event that is to be processed by the agent runtime in the execution context object identified by the execution context value stored in the execution context field;
• a time out field for storing a time out value that represents an amount of time after which the attempt to asynchronously invoke an agent runtime will time out;
• a callback field for storing a callback value that represents an object to be called when the asynchronous invocation of the agent runtime is complete;
• a state field for storing a state value that distinguishes the asynchronous invocation of the agent runtime from other invocations of the agent runtime;
• and
• a contexts field for storing a list of context values that represent a corresponding list of objects from a data processing pipeline that can be accessed from the execution context object identified by the execution context value stored in the execution context field;
• divert pipeline data related to the represented messaging command out of the data pipeline to the identified agent to extend the functionality of the data pipeline subsequent to loading the agent;
• and
• inject results from the identified agent back into the data pipeline for further processing.

At a computer system, the computer system including a processor and system memory, the computer system also including a data pipeline comprising a sequential arrangement of a plurality of pipeline modules, an electronic message received at the data pipeline being sequential processed by pipeline modules such that the output of at least one pipeline module is the input to the next sequential pipeline module, a method for invoking an agent runtime to customize the functionality of a data pipeline, the method comprising:

• receiving electronic message related data at a pipeline module in the sequential arrangement of pipeline modules;
• detecting the occurrence of an event at the pipeline module in response to processing the electronic message related data at the pipeline module;
• mapping the detected event to a specified topic string representing an electronic messaging command;
• scanning configuration data for stored agents to attempt to identify any stored agents having topic strings corresponding to the specified topic string;
• indentifying an agent having a topic string corresponding to the specified topic string, the agent having the corresponding topic string indicating that the agent is self-configured to hook into the detected event;
• the processor loading the identified agent such that the identified agent can be executed to manipulate messaging data associated with the detected event in accordance with the configuration data for the identified agent, loading the identified agent including supplying appropriate values to an invoke API to invoke the identified agent, supplying appropriate values including supplying values for:
• an execution context field for storing an execution context value that identifies an execution context object for an agent runtime;
• an event topic field for storing an event topic value that represents a type of event that is to be processed by the agent runtime in the execution context object identified by the execution context value stored in the execution context field;
• a contexts field for storing a list of context values that represent a corresponding list of objects from a data processing pipeline that can be accessed from the execution context object identified by the execution context value stored in the execution context field;
• diverting pipeline data related to the represented messaging command out of the data pipeline to the identified agent to extend the functionality of the data pipeline subsequent to loading the agent;
• and
• injecting results from the identified agent back into the data pipeline for further processing.

At a computer system, the computer system including a processor and system memory, the computer system also including a data pipeline comprising a sequential arrangement of a plurality of pipeline modules, an electronic message received at the data pipeline being sequential processed by pipeline modules such that the output of at least one pipeline module is the input to the next sequential pipeline module, a method for invoking an agent runtime to customize the functionality of a data pipeline, the method comprising:

• receiving electronic message related data at a pipeline module in the sequential arrangement of pipeline modules;
• detecting the occurrence of an event at the pipeline module in response to processing the electronic message related data at the pipeline module;
• mapping the detected event to a specified topic string representing an electronic messaging command;
• scanning configuration data for stored agents to attempt to identify any stored agents having topic strings corresponding to the specified topic string;
• indentifying an agent having a topic string corresponding to the specified topic string, the agent having the corresponding topic string indicating that the agent is self-configured to hook into the detected event;
• the processor asynchronously invoking the identified agent such that the identified agent can be executed to manipulate data related to the detected event in accordance with the configuration data for the identified agent, asynchronously invoking the identified agent including supplying appropriate values to an invoke API for the identified agent, supplying appropriate values including supplying values for:
• an execution context field for storing an execution context value that identifies an execution context object for an agent runtime;
• an event topic field for storing an event topic value that represents a type of event that is to be processed by the agent runtime in the execution context object identified by the execution context value stored in the execution context field;
• a time out field for storing a time out value that represents an amount of time after which the attempt to asynchronously invoke an agent runtime will time out;
• a callback field for storing a callback value that represents an object to be called when the asynchronous invocation of the agent runtime is complete;
• a state field for storing a state value that distinguishes the asynchronous invocation of the agent runtime from other invocations of the agent runtime;
• and
• a contexts field for storing a list of context values that represent a corresponding list of objects from a data processing pipeline that can be accessed from the execution context object identified by the execution context value stored in the execution context field;
• diverting pipeline data related to the represented messaging command out of the data pipeline to the identified agent to extend the functionality of the data pipeline subsequent to loading the agent;
• and
• injecting results from the identified agent back into the data pipeline for further processing.