157 lines
4.0 KiB
Gherkin
157 lines
4.0 KiB
Gherkin
Feature: Reactive Stream Processing
|
|
As a developer using CleverAgents
|
|
I want to create and manage reactive streams
|
|
So that I can build complex agent workflows with RxPy
|
|
|
|
Background:
|
|
Given the CleverAgents reactive system is available
|
|
|
|
Scenario: Create a basic cold stream
|
|
Given I have a stream configuration:
|
|
"""
|
|
{
|
|
"name": "test_stream",
|
|
"type": "cold",
|
|
"operators": []
|
|
}
|
|
"""
|
|
When I create the reactive stream
|
|
Then the stream should be created successfully
|
|
And the stream type should be "cold"
|
|
|
|
Scenario: Create a hot stream with operators
|
|
Given I have a stream configuration:
|
|
"""
|
|
{
|
|
"name": "hot_stream",
|
|
"type": "hot",
|
|
"operators": [
|
|
{
|
|
"type": "map",
|
|
"params": {
|
|
"transform": {
|
|
"type": "format",
|
|
"template": "Processed: {content}"
|
|
}
|
|
}
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
When I create the reactive stream
|
|
Then the stream should be created successfully
|
|
And the stream type should be "hot"
|
|
And the stream should have 1 operator
|
|
|
|
Scenario: Stream with debounce operator
|
|
Given I have a stream configuration:
|
|
"""
|
|
{
|
|
"name": "debounced_stream",
|
|
"type": "cold",
|
|
"operators": [
|
|
{
|
|
"type": "debounce",
|
|
"params": {
|
|
"duration": 1.0
|
|
}
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
When I create the reactive stream
|
|
And I send messages to the stream rapidly
|
|
Then only the last message should be processed
|
|
|
|
Scenario: Stream with filter operator
|
|
Given I have a stream configuration:
|
|
"""
|
|
{
|
|
"name": "filtered_stream",
|
|
"type": "cold",
|
|
"operators": [
|
|
{
|
|
"type": "filter",
|
|
"params": {
|
|
"condition": {
|
|
"type": "content_contains",
|
|
"text": "important"
|
|
}
|
|
}
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
When I create the reactive stream
|
|
And I send a message "important information" to the stream
|
|
And I send a message "regular message" to the stream
|
|
Then only messages containing "important" should be processed
|
|
|
|
Scenario: Stream with buffer operator
|
|
Given I have a stream configuration:
|
|
"""
|
|
{
|
|
"name": "buffered_stream",
|
|
"type": "cold",
|
|
"operators": [
|
|
{
|
|
"type": "buffer",
|
|
"params": {
|
|
"count": 3
|
|
}
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
When I create the reactive stream
|
|
And I send 5 messages to the stream
|
|
Then messages should be processed in batches of 3
|
|
|
|
Scenario: Merge multiple streams
|
|
Given I have streams "stream1", "stream2", and "stream3"
|
|
When I merge them into "merged_stream"
|
|
Then the merged stream should receive messages from all source streams
|
|
|
|
Scenario: Split stream based on content
|
|
Given I have a stream "input_stream"
|
|
When I split it with conditions:
|
|
"""
|
|
{
|
|
"questions": {
|
|
"type": "content_contains",
|
|
"text": "?"
|
|
},
|
|
"commands": {
|
|
"type": "content_contains",
|
|
"text": "execute"
|
|
}
|
|
}
|
|
"""
|
|
Then messages with "?" should go to "questions" stream
|
|
And messages with "execute" should go to "commands" stream
|
|
|
|
Scenario: Stream error handling with retry
|
|
Given I have a stream with retry operator:
|
|
"""
|
|
{
|
|
"name": "retry_stream",
|
|
"type": "cold",
|
|
"operators": [
|
|
{
|
|
"type": "map",
|
|
"params": {
|
|
"function": "failing_function"
|
|
}
|
|
},
|
|
{
|
|
"type": "retry",
|
|
"params": {
|
|
"count": 3
|
|
}
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
When I send a message that causes an error
|
|
Then the operation should be retried 3 times
|
|
And eventually succeed or fail definitively |