integration_core 1.0.2

class: integration_core_module

[8:14] static extends: object

Module JAR loader class.

Methods

• integration_core_module ()

class: ScatterGather

[258:14] (extern: com.lehman.aussom.ScatterGather) extends: object

ScatterGather sends the same payload to several recipient callbacks across a bounded thread pool and blocks until every recipient has finished. Each recipient's return value becomes its result, gathered in recipient order.

Methods

• ScatterGather (Payload, list Recipients)

  Creates a new ScatterGather.

  • @p Payload is the value handed to every recipient.
  • @p Recipients is a list of callbacks, each invoked once with the payload.

• init (Payload, list Recipients)

• setNumThreads (int NumThreads)

  Sets the maximum number of worker threads. Default: the number of available processors.

  • @p NumThreads is an int with the pool size.
  • @r this object

• setThreadTimeout (int ThreadTimeoutMills)

  Sets the per-worker timeout in milliseconds. 0 means no timeout. Default: 0.

  • @p ThreadTimeoutMills is an int with the timeout in milliseconds.
  • @r this object

• setReturnValues (bool ReturnValues = true)

  Sets whether the run collects and returns the recipient return values. Default: true.

  • @p ReturnValues is a bool with the flag.
  • @r this object

• run ()

  Sends the payload to every recipient in parallel and blocks until done.

  • @r A list of return values aligned to recipient order, with null for any recipient that failed or timed out, or null when return values are disabled.

class: IdempotentReceiver

[399:7] extends: object

IdempotentReceiver is a thread-safe duplicate filter. receive(Id) returns true the first time an id is seen and false for every repeat, so the caller can process a new message once and drop the duplicates. Seen ids are kept in a mutable in-memory map by default, or in a JDBC table when a Jdbc is set, for a store that survives a restart and can be shared. A single-permit Semaphore from the concurrent library makes the check-and-record one atomic step, so receive is safe to call from many threads at once.

Members

• seen

  In-memory set of seen ids (a map used as a set).

• db

  Connected Jdbc object for the persistent backend, or null for in-memory.

• tableName

  Table name for the JDBC backend.

• mutex

  Single-permit Semaphore used as a mutex around the critical section.

• lockWaitMs

  How long receive waits for the mutex before failing, in milliseconds.

Methods

• IdempotentReceiver ()

  Constructs a receiver using the in-memory backend.

• setJdbc (Db)

  Switches to the JDBC backend, using the provided connected Jdbc object.

  • @p Db is a connected Jdbc object.
  • @r this object

• setTableName (string TableName)

  Sets the table name for the JDBC backend. Default: idempotent_receiver.

  • @p TableName is the table name string.
  • @r this object

• setup ()

  Creates the id table if it does not already exist. JDBC backend only; a no-op when using the in-memory backend.

  • @r this object

• receive (string Id)

  Records the id and reports whether it is new. Returns true the first time an id is seen and false for every repeat. Atomic and safe to call from many threads at once. Blocks waiting for the internal mutex and throws if it cannot be acquired in time.

  • @p Id is the message id string.
  • @r true if the id is new, false if it is a duplicate.

• isDuplicateKeyError (Err)

  Reports whether a caught database error is a primary-key/unique constraint violation, which the receiver reads as "this id was already seen."

  • @p Err is the caught exception.
  • @r true if the error is a duplicate-key violation.

class: IMessage

[26:7] extends: object

IMessage is the standard message type passed between the EIP tools. It holds message metadata, the content type, the body, and a correlation id. It is a plain Aussom value type so it works in both the Aussom CLI and Aussom Server.

Members

• attributes

  Aussom map of message metadata (headers, query params, and so on).

• type

  Content type string describing the body (for example application/json).

• body

  The message body: a Buffer for binary, a string for text/JSON/XML, or an Aussom object.

• correlationId

  Correlation id string tying related messages together.

Methods

• IMessage ()

  Constructs a new IMessage with an empty attribute map and a freshly generated correlation id.

• getAttributes ()

  Gets the attribute map.

  • @r The attribute map.

• setAttributes (map Attributes)

  Sets the attribute map.

  • @p Attributes is the map of message metadata.
  • @r this object

• getType ()

  Gets the content type.

  • @r A string with the content type.

• setType (string Type)

  Sets the content type.

  • @p Type is a string with the content type.
  • @r this object

• getBody ()

  Gets the message body.

  • @r The message body.

• setBody (Body)

  Sets the message body.

  • @p Body is the message body.
  • @r this object

• getCorrelationId ()

  Gets the correlation id.

  • @r A string with the correlation id.

• setCorrelationId (string CorrelationId)

  Sets the correlation id.

  • @p CorrelationId is a string with the correlation id.
  • @r this object

• copy ()

  Returns a deep copy of this message. The attribute map, the body, and any nested maps and lists are copied so the copy is independent of the original. Strings and numbers are immutable and shared safely. Other object values (such as a Buffer) are copied by reference. The copy keeps the same correlation id.

  • @r A new IMessage that is a deep copy of this one.

• deepCopy (Val)

  Recursively deep copies maps and lists. Other values are returned as-is.

  • @p Val is the value to copy.
  • @r A deep copy of the value.

class: ParallelForeach

[210:14] (extern: com.lehman.aussom.ParallelForeach) extends: object

ParallelForeach runs a callback over each value in a list across a bounded thread pool and blocks until every value has been processed. The callback's return value becomes that value's result.

Methods

• ParallelForeach (list Values, callback OnExecute)

  Creates a new ParallelForeach.

  • @p Values is a list of values to process.
  • @p OnExecute is the callback invoked once per value, receiving the value.

• init (list Values, callback OnExecute)

• setNumThreads (int NumThreads)

  Sets the maximum number of worker threads. Default: the number of available processors.

  • @p NumThreads is an int with the pool size.
  • @r this object

• setThreadTimeout (int ThreadTimeoutMills)

  Sets the per-worker timeout in milliseconds. 0 means no timeout. Default: 0.

  • @p ThreadTimeoutMills is an int with the timeout in milliseconds.
  • @r this object

• setReturnValues (bool ReturnValues = true)

  Sets whether the run collects and returns the callback return values. Default: true.

  • @p ReturnValues is a bool with the flag.
  • @r this object

• run ()

  Runs the callback over every value in parallel and blocks until done.

  • @r A list of return values aligned to input order, with null for any value that failed or timed out, or null when return values are disabled.

class: Pipeline

[307:7] extends: object

Pipeline runs a value through an ordered list of step callbacks, each step's output feeding the next. A step returns the next value to continue, or null to drop the value and stop the chain. It is the sequential sibling of ParallelForeach and is pure Aussom, since it is a simple loop over the steps.

Members

• steps

  Ordered list of step callbacks.

• onStep

  Optional per-step observability hook.

• onError

  Optional error handler.

Methods

• Pipeline (list Steps)

  Constructs a Pipeline with the ordered step callbacks.

  • @p Steps is a list of step callbacks (may be empty).

• add (callback Step)

  Appends a step to the chain.

  • @p Step is the step callback to append.
  • @r this object

• setOnStep (callback OnStep)

  Sets the per-step observability hook. The hook is invoked after each step as OnStep.call(StepIndex, Message) with the value that step produced.

  • @p OnStep is the hook callback.
  • @r this object

• setOnError (callback OnError)

  Sets the error handler. When set, a step that throws is caught and the handler is invoked as OnError.call(StepIndex, Message, Error), where Message is the input to the failing step. When not set, a thrown error propagates out of run.

  • @p OnError is the error handler callback.
  • @r this object

• run (Message)

  Runs the value through every step in order and returns the final result. A step returning null stops the chain and run returns null. With an error handler set, a thrown error is routed to it and run returns null; otherwise the error propagates.

  • @p Message is the value to run through the steps.
  • @r The final value, or null if a step dropped it or a handled error stopped it.

class: IntegrationCore

[17:14] (extern: com.lehman.aussom.IntegrationCore) extends: object

This extern class is a link to the .java file.

Methods

• aussomFunction (string Variable)

class: iutils

[138:14] static extends: object

iutils provides static helper functions for the EIP tools.

Methods

• newCorrelationId ()

  Generates a new correlation id string.

  • @r A string with a newly generated unique id.

• message (Req)

  Converts an Aussom Server HttpReq object into an IMessage. The request headers, method, path, and query parameters are copied into the message attributes. The correlation id is taken from the X-Correlation-ID header when present or generated otherwise. Header lookups ignore case. The body and type depend on the Content-Type header: - For form requests (multipart/form-data or application/x-www-form-urlencoded), the parsed form data is used as the body. This is a native Aussom map of the fields, with any uploaded files as Buffer values, so the type is set to application/aussom. The original Content-Type stays available in the attribute headers. - Otherwise the body is read as a string and the type is the Content-Type header, defaulting to text/plain when absent. Only the HttpReq accessor methods are used, so the resulting IMessage holds native Aussom values only and stays usable outside the server.

  • @p Req is an Aussom Server HttpReq object.
  • @r A new IMessage built from the request.