菜单
Provides expertise in the message package architecture for building CloudEvents-based pipelines in gopipe. Apply when working with the message package, Engine, Router, Handler, Matcher, or designing event-driven systems.
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
Add an entry to the CHANGELOG under [Unreleased]. Usage: /changelog TYPE DESC (e.g. /changelog Added support for Redis adapter)
Create a new Architecture Decision Record. Usage: /create-adr TITLE (e.g. /create-adr add-redis-adapter)
Create a new feature branch from develop. Usage: /create-feature NAME (e.g. /create-feature message-validation)
Create a new implementation plan document. Usage: /create-plan TITLE (e.g. /create-plan redis-adapter)
Provides Go development standards and best practices for the gopipe codebase. Apply when writing or reviewing Go code: testing patterns, godoc, error handling, API conventions, and common anti-patterns to avoid.
Check documentation quality and consistency across the project.
| name | building-message-pipelines |
| description | Provides expertise in the message package architecture for building CloudEvents-based pipelines in gopipe. Apply when working with the message package, Engine, Router, Handler, Matcher, or designing event-driven systems. |
| user-invocable | false |
RawInputs → Unmarshal ─┐
├→ Merger → Router → Distributor
TypedInputs ───────────┘ │
┌──────────┴──────────┐
TypedOutput RawOutput
Each raw input has its own unmarshal pipe feeding into a single shared merger.
engine := message.NewEngine(message.EngineConfig{
Marshaler: message.NewJSONMarshaler(),
ShutdownTimeout: 5 * time.Second,
})
engine.AddHandler("handler-name", matcher, message.NewCommandHandler(
func(ctx context.Context, cmd InputType) ([]OutputType, error) {
return []OutputType{{...}}, nil
},
message.CommandHandlerConfig{
Source: "/source",
Naming: message.DotNaming,
},
))
type Handler interface {
EventType() string // CE type for routing (e.g., "order.created")
NewInput() any // Creates instance for unmarshaling
Handle(ctx context.Context, msg *Message) ([]*Message, error)
}
type Matcher interface {
Match(attrs Attributes) bool // Attributes only — not *Message
}
Operates on Attributes only (not *Message) to avoid wrapper allocation for raw messages.
// Raw input ([]byte data)
input := make(chan *message.RawMessage, 10)
engine.AddRawInput("name", matcher, input)
// Raw output
output, _ := engine.AddRawOutput("name", matcher)
// Typed input
typedInput := make(chan *message.Message, 10)
engine.AddInput("name", matcher, typedInput)
// Typed output
typedOutput, _ := engine.AddOutput("name", matcher)
Loopback is NOT built into Engine. Use plugin.Loopback:
engine.AddPlugin(plugin.Loopback("step1-loop", &typeMatcher{"step1.completed"}))
| Naming | Output | Example |
|---|---|---|
DotNaming | type.subtype | order.created |
KebabNaming | type-subtype | order-created |
SnakeNaming | type_subtype | order_created |
ctx, cancel := context.WithCancel(context.Background())
done, _ := engine.Start(ctx)
close(input) // Close inputs first
cancel() // Then cancel context
<-done // Wait for shutdown
Combined Marshaler with Registry — rejected: single responsibility. Marshaler is pure serialization; Handler.NewInput() provides instances.
PipeHandler Interface — rejected: over-engineering. EventType() returning "*" for multi-type is a hack.
Named Outputs with RouteOutput — rejected: pattern matching on CE type is more declarative.