Layer 2 is Priostack's embedded EIP integration layer — 9 Enterprise Integration Patterns running inside your BPM runtime. No Kafka, no RabbitMQ, no ops overhead.
EIP Primitives
Every pattern ships as a typed Go struct. No XML, no DSL files, no external bus — just import the package and wire your workflow.
Point-to-point and pub-sub message delivery between components
Send and receive messages with correlation key support
Route messages to different channels via FEEL expressions
Drop or queue messages that don't match a predicate
Collect related messages and fire a callback when a batch completes
Split a list variable into parallel messages for fan-out
Map and transform message fields using FEEL expressions
Chain filters in sequence — Pipeline.Run(ctx, msg)
First-match-wins deduplication with configurable TTL
Reliability
Choose the right deduplication strategy for your latency and persistence requirements. All five work without an external broker.
| Mechanism | Scope | Storage | When to use |
|---|---|---|---|
| CorrelationContext | Per-channel | In-memory | High-speed dedup within a session |
| MessageID check | Global | Persistent | Cross-restart exactly-once |
| Database upsert | Application | PostgreSQL | Business-level idempotency |
| Conditional routing | Pipeline | None | Skip duplicates at router |
| Acknowledgement timeout | Endpoint | In-memory | At-least-once with dedup |
Code Examples
Every primitive is a Go struct — no config files, no annotations, no framework lock-in.
// Create a pub-sub channel for the credit-check stage ch := layer2.NewMessageChannel("credit-check", layer2.PubSub) ch.Publish(ctx, layer2.Message{Key: loanID, Body: applicant}) ch.Subscribe(func(msg layer2.Message) { result := creditService.Evaluate(msg.Body.(Applicant)) msg.Reply(result) })
// Aggregate 100 loan applications before risk assessment agg := layer2.NewAggregator("loan-batch", layer2.CorrelationKey("portfolio_id"), layer2.CompletionCondition("count(messages) >= 100"), layer2.OnComplete(func(batch []layer2.Message) { risk := riskEngine.AssessBatch(batch) downstream.Publish(ctx, risk) }), )
// Dedup incoming messages, then run the credit pipeline ctx := layer2.NewCorrelationContext( layer2.TTL(5 * time.Minute), layer2.FirstMatchWins(), ) pipeline := layer2.NewPipeline( ctx.Filter(), // drop duplicates creditFilter, // business logic translatorStep, // transform output ) pipeline.Run(ctx, incomingMsg)
Competitive Analysis
External brokers solve hard distribution problems — but most enterprise workflows don't need them. Layer 2 removes the accidental complexity.
Kafka requires 3+ brokers, ZooKeeper or KRaft, topic management, consumer group coordination, and a dedicated ops team. It solves large-scale log distribution — not in-process workflow messaging.
Layer 2: embedded, zero config, starts in <1 msRabbitMQ needs a dedicated server, an AMQP client library, exchange and binding configuration, and a connection string in every service. Infrastructure overhead precedes the first message.
Layer 2: import one package, no infrastructureSQS charges per API call, adds 50–200 ms of network latency on every message, and requires AWS credentials, SDK configuration, and IAM policies before you can enqueue anything.
Layer 2: sub-millisecond, zero cost per messageQuick Start
No YAML manifests. No Docker containers. No service mesh. Just Go code that compiles and runs.
Instantiate a named channel in the channel registry — point-to-point or pub-sub.
Call ch.Publish(ctx, msg) from any BPMN service task output handler.
Register a subscriber function. It receives typed messages and can reply inline.
Wrap your pipeline with a CorrelationContext to drop duplicate messages before they reach business logic.
Templates
Deploy a fully wired integration pattern into your workspace in one click. Each template ships with a BPMN diagram, Go worker, and unit test.
Collect loan applications by portfolio ID and trigger a risk batch once the batch size threshold is met.
Deploy template →Route loan decisions to low-risk auto-approval, manual review, or hard-decline channels via a FEEL expression.
Deploy template →Drop duplicate webhook events within a configurable TTL window before they enter the processing pipeline.
Deploy template →Broadcast a process completion event to audit, email, and analytics subscribers simultaneously.
Deploy template →Chain a FEEL-based field mapper and a schema validator in a single pipeline stage before downstream delivery.
Deploy template →FAQ
If your question isn't here, reach out to the team via the console or the community forum.
Yes. Layer 2 is a standalone Go package. You can import layer2 independently of the BPMN runtime and use message channels, pipelines, and aggregators in any Go application.
No. By default, all message channels run in-process using Go channels and sync primitives. There is no TCP socket, no serialization overhead, and no external dependency to deploy or manage.
External broker adapters are on the roadmap. The planned API will let you swap the transport backend of any MessageChannel from in-process to Kafka or AMQP without changing application code.
A PostgreSQL-backed channel store is currently on the roadmap. In the interim, use the MessageID check mechanism against a persistent store to achieve cross-restart exactly-once semantics in your worker.
In internal benchmarks, a 5-stage pipeline processes 10,000 messages per second on a single core with p99 latency under 200 microseconds. Results vary with payload size and filter complexity.