Struct PlatformSource 

pub struct PlatformSource { /* private fields */ }

Platform source that reads events from Redis Streams.

This source connects to a Redis instance and consumes CloudEvent-wrapped messages from a stream using consumer groups. It supports both data events (node/relation changes) and control events (query subscriptions).

Fields

• base: Common source functionality (dispatchers, status, lifecycle)
• config: Platform-specific configuration (Redis connection, stream settings)

Implementations

impl PlatformSource

pub fn builder(id: impl Into<String>) -> PlatformSourceBuilder

Create a builder for PlatformSource

Example

use drasi_source_platform::PlatformSource;

let source = PlatformSource::builder("my-platform-source")
    .with_redis_url("redis://localhost:6379")
    .with_stream_key("my-changes")
    .with_bootstrap_provider(my_provider)
    .build()?;

pub fn new(id: impl Into<String>, config: PlatformSourceConfig) -> Result<Self>

Create a new platform source.

The event channel is automatically injected when the source is added to DrasiLib via add_source().

Arguments

• id - Unique identifier for this source instance
• config - Platform source configuration

Returns

A new PlatformSource instance, or an error if construction fails.

Errors

Returns an error if the base source cannot be initialized.

Example

use drasi_source_platform::{PlatformSource, PlatformSourceBuilder};

let config = PlatformSourceBuilder::new("my-platform-source")
    .with_redis_url("redis://localhost:6379")
    .with_stream_key("my-changes")
    .build()?;

impl PlatformSource

pub fn test_subscribe(&self) -> Box<dyn ChangeReceiver<SourceEventWrapper>>

Create a test subscription to this source (synchronous)

This method delegates to SourceBase and is provided for convenience in tests. Note: Use test_subscribe_async() in async contexts to avoid runtime issues.

pub async fn test_subscribe_async( &self, ) -> Box<dyn ChangeReceiver<SourceEventWrapper>>

Create a test subscription to this source (async)

This method delegates to SourceBase and is provided for convenience in async tests. Prefer this method over test_subscribe() in async contexts.

Trait Implementations

impl Source for PlatformSource

fn id(&self) -> &str

Get the source’s unique identifier

fn type_name(&self) -> &str

Get the source type name (e.g., “postgres”, “http”, “mock”)

fn properties(&self) -> HashMap<String, Value>

Get the source’s configuration properties for inspection

fn auto_start(&self) -> bool

Whether this source should auto-start when DrasiLib starts

fn start<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Start the source

fn stop<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Stop the source

fn status<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ComponentStatus> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get the current status of the source

fn subscribe<'life0, 'async_trait>( &'life0 self, settings: SourceSubscriptionSettings, ) -> Pin<Box<dyn Future<Output = Result<SubscriptionResponse>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to this source for change events

fn as_any(&self) -> &dyn Any

Downcast helper for testing - allows access to concrete types

fn initialize<'life0, 'async_trait>( &'life0 self, context: SourceRuntimeContext, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Initialize the source with runtime context.

fn set_bootstrap_provider<'life0, 'async_trait>( &'life0 self, provider: Box<dyn BootstrapProvider + 'static>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Set the bootstrap provider for this source

fn dispatch_mode(&self) -> DispatchMode

Get the dispatch mode for this source (Channel or Broadcast)

fn supports_replay(&self) -> bool

Whether this source supports positional replay via resume_from.

fn deprovision<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Permanently clean up internal state when the source is being removed.