runtime

package

Go to main page

Versions in this module

v0

v0.0.196

Dec 11, 2024

Changes in this version

type ServiceApp

+ func (app *ServiceApp) ServiceContext() interface{}

v0.0.195

Dec 11, 2024

v0.0.194

Dec 11, 2024

v0.0.193

Dec 11, 2024

v0.0.192

Dec 11, 2024

Changes in this version

type ServiceApp

+ func (app *ServiceApp) AddHandler(pattern string, handler http.Handler)

+ func (app *ServiceApp) RouteHandler() http.Handler

v0.0.191

Dec 11, 2024

v0.0.190

Dec 10, 2024

Changes in this version

+ type SinkFunction interface

+ Sink func(Stream, T, error, Collect[R])

+ type SinkFunctionContext struct

v0.0.189

Dec 10, 2024

v0.0.188

Dec 10, 2024

v0.0.187

Dec 9, 2024

v0.0.186

Dec 9, 2024

Changes in this version

+ type SinkCallback interface

+ Done func(T, error)

+ type SinkConsumer interface

+ SetSinkCallback func(SinkCallback[T])

+ type SinkErrorFunction interface

+ SinkError func(Stream, T, error, Collect[R])

+ type SinkErrorFunctionContext struct

type SinkStream

+ func (s *SinkStream[T, R]) Done(value T, err error)

+ func (s *SinkStream[T, R]) SetSinkConsumer(sinkConsumer SinkConsumer[T])

type TypedSinkStream

+ SetSinkConsumer func(SinkConsumer[T])

v0.0.184

Dec 2, 2024

Changes in this version

type ServiceApp

+ func (app *ServiceApp) GetHttpRoute() httproute.HttpRoute

+ func (app *ServiceApp) ServiceInit() error

type ServiceExecutionEnvironment

+ GetHttpRoute func() httproute.HttpRoute

+ ServiceInit func() error

v0.0.180

Nov 24, 2024

Changes in this version

+ type AppInputStream struct

+ func MakeAppInputStream[T any](name string, env ServiceExecutionEnvironment) *AppInputStream[T]

+ func (s *AppInputStream[T]) Consume(value T)

type ServiceApp

+ func (app *ServiceApp) GetPriorityTaskPool(name string) pool.PriorityTaskPool

+ func (app *ServiceApp) GetTaskPool(name string) pool.TaskPool

type ServiceExecutionEnvironment

+ GetPriorityTaskPool func(name string) pool.PriorityTaskPool

+ GetTaskPool func(name string) pool.TaskPool

v0.0.177

Nov 13, 2024

v0.0.176

Nov 11, 2024

Changes in this version

type ServiceApp

+ func (app *ServiceApp) ServiceDependency() environment.ServiceDependency

v0.0.175

Nov 11, 2024

Changes in this version

type DataSinkEndpointConsumer

+ func (ec *DataSinkEndpointConsumer[T]) Stream() TypedSinkStream[T]

type DataSourceEndpointConsumer

+ func (ec *DataSourceEndpointConsumer[T]) Stream() TypedInputStream[T]

v0.0.174

Nov 9, 2024

v0.0.173

Nov 9, 2024

Changes in this version

+ type Collection struct

+ func NewCollection[T any](data []T) Collection[T]

+ func (s Collection[T]) At(i int) T

+ func (s Collection[T]) Len() int

v0.0.172

Nov 8, 2024

v0.0.171

Nov 6, 2024

v0.0.170

Nov 6, 2024

Changes in this version

type OutStubBinaryKVStream

+ func (s *OutStubBinaryKVStream[T]) GetConsumers() []Stream

type OutStubBinaryStream

+ func (s *OutStubBinaryStream[T]) GetConsumers() []Stream

type OutStubStream

+ func (s *OutStubStream[T]) GetConsumers() []Stream

v0.0.168

Nov 6, 2024

v0.0.167

Nov 5, 2024

Changes in this version

type JoinLink

+ func (s *JoinLink[K, T1, T2, R]) Validate() error

type MergeLink

+ func (s *MergeLink[T]) Validate() error

type MultiJoinLinkStream

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) Validate() error

type ServiceStream

+ func (s *ServiceStream[T]) Validate() error

type SplitLink

+ func (s *SplitLink[T]) Validate() error

type SplitStream

+ func (s *SplitStream[T]) Validate() error

type Stream

+ Validate func() error

v0.0.166

Nov 5, 2024

v0.0.164

Nov 4, 2024

v0.0.163

Nov 3, 2024

Changes in this version

type ServiceStream

+ func (s *ServiceStream[T]) GetConfig() *config.StreamConfig

+ func (s *ServiceStream[T]) GetEnvironment() ServiceExecutionEnvironment

+ func (s *ServiceStream[T]) GetId() int

+ func (s *ServiceStream[T]) GetName() string

+ func (s *ServiceStream[T]) GetTransformationName() string

+ func (s *ServiceStream[T]) GetTypeName() string

v0.0.162

Oct 29, 2024

v0.0.161

Oct 29, 2024

Changes in this version

type ServiceApp

+ func (app *ServiceApp) AppConfig() *config.ServiceAppConfig

+ func (app *ServiceApp) Log() log.Logger

+ func (app *ServiceApp) Metrics() metrics.Metrics

+ func (app *ServiceApp) Release()

+ func (app *ServiceApp) ServiceConfig() *config.ServiceConfig

type ServiceExecutionEnvironment

+ Release func()

v0.0.158

Oct 28, 2024

v0.0.156

Oct 28, 2024

Changes in this version

type ServiceApp

+ func (app *ServiceApp) GetLog() log.Logger

v0.0.154

Oct 27, 2024

v0.0.153

Oct 27, 2024

v0.0.152

Oct 27, 2024

v0.0.151

Oct 27, 2024

v0.0.150

Oct 26, 2024

Changes in this version

type JoinStream

+ func (s *JoinStream[K, T1, T2, R]) GetJoinStorageType() api.JoinStorageType

+ func (s *JoinStream[K, T1, T2, R]) GetMetrics() metrics.Metrics

+ func (s *JoinStream[K, T1, T2, R]) GetRenewTTL() bool

+ func (s *JoinStream[K, T1, T2, R]) GetServiceName() string

+ func (s *JoinStream[K, T1, T2, R]) GetTTL() time.Duration

type MultiJoinStream

+ func (s *MultiJoinStream[K, T, R]) GetJoinStorageType() api.JoinStorageType

+ func (s *MultiJoinStream[K, T, R]) GetMetrics() metrics.Metrics

+ func (s *MultiJoinStream[K, T, R]) GetRenewTTL() bool

+ func (s *MultiJoinStream[K, T, R]) GetServiceName() string

+ func (s *MultiJoinStream[K, T, R]) GetTTL() time.Duration

v0.0.149

Oct 26, 2024

v0.0.148

Oct 25, 2024

v0.0.147

Oct 25, 2024

v0.0.146

Oct 25, 2024

Changes in this version

type ServiceApp

+ func (app *ServiceApp) GetAppConfig() *config.ServiceAppConfig

v0.0.145

Oct 25, 2024

Changes in this version

+ type ServiceLoader interface

+ Stop func()

v0.0.144

Oct 21, 2024

v0.0.142

Oct 19, 2024

v0.0.141

Oct 15, 2024

v0.0.140

Oct 12, 2024

v0.0.138

Oct 11, 2024

v0.0.137

Oct 8, 2024

v0.0.136

Oct 8, 2024

v0.0.135

Oct 8, 2024

v0.0.134

Oct 7, 2024

Changes in this version

type AppSinkStream

+ func (s *AppSinkStream[T]) GetConsumers() []Stream

type ConsumedStream

+ func (s *ConsumedStream[T]) GetConsumers() []Stream

type DataSink

+ GetEnvironment func() ServiceExecutionEnvironment

type DataSinkEndpoint

+ func (ep *DataSinkEndpoint) GetEnvironment() ServiceExecutionEnvironment

type DataSource

+ GetEnvironment func() ServiceExecutionEnvironment

type DataSourceEndpoint

+ func (ep *DataSourceEndpoint) GetEnvironment() ServiceExecutionEnvironment

type InStubKVStream

+ func (s *InStubKVStream[T]) Consume(value T)

type InStubStream

+ func (s *InStubStream[T]) Consume(value T)

type InputDataSource

+ func (ds *InputDataSource) GetEnvironment() ServiceExecutionEnvironment

type InputEndpoint

+ GetEnvironment func() ServiceExecutionEnvironment

type InputStream

+ func (s *InputStream[T]) Consume(value T)

type JoinLink

+ func (s *JoinLink[K, T1, T2, R]) GetConsumers() []Stream

+ func (s *JoinLink[K, T1, T2, R]) GetEnvironment() ServiceExecutionEnvironment

type MergeLink

+ func (s *MergeLink[T]) GetConsumers() []Stream

+ func (s *MergeLink[T]) GetEnvironment() ServiceExecutionEnvironment

type MergeStream

+ func (s *MergeStream[T]) Consume(value T)

type MultiJoinLinkStream

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetConsumers() []Stream

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetEnvironment() ServiceExecutionEnvironment

type OutputDataSink

+ func (ds *OutputDataSink) GetEnvironment() ServiceExecutionEnvironment

type ServiceApp

+ func (app *ServiceApp) GetRuntime() ServiceExecutionRuntime

+ type ServiceExecutionEnvironment interface

+ AddDataSink func(dataSink DataSink)

+ AddDataSource func(dataSource DataSource)

+ Delay func(duration time.Duration, f func())

+ GetConsumeTimeout func(from int, to int) time.Duration

+ GetDataSink func(id int) DataSink

+ GetDataSource func(id int) DataSource

+ GetEndpointReader func(endpoint Endpoint, stream Stream, valueType reflect.Type) EndpointReader

+ GetEndpointWriter func(endpoint Endpoint, stream Stream, valueType reflect.Type) EndpointWriter

+ GetMetrics func() metrics.Metrics

+ GetRuntime func() ServiceExecutionRuntime

+ GetSerde func(valueType reflect.Type) (serde.Serializer, error)

+ SetConfig func(config config.Config)

+ Start func(context.Context) error

+ Stop func(context.Context)

+ StreamsInit func(ctx context.Context)

+ type ServiceExecutionRuntime interface

type SinkEndpoint

+ GetEnvironment func() ServiceExecutionEnvironment

type SinkStream

+ func (s *SinkStream[T]) GetConsumers() []Stream

type SplitLink

+ func (s *SplitLink[T]) GetConsumers() []Stream

+ func (s *SplitLink[T]) GetEnvironment() ServiceExecutionEnvironment

type SplitStream

+ func (s *SplitStream[T]) GetConsumers() []Stream

type Stream

+ GetConsumers func() []Stream

+ GetEnvironment func() ServiceExecutionEnvironment

type StreamBase

+ func (s *StreamBase[T]) GetEnvironment() ServiceExecutionEnvironment

v0.0.132

Sep 26, 2024

v0.0.131

Sep 13, 2024

v0.0.130

Sep 13, 2024

v0.0.129

Sep 9, 2024

v0.0.126

Sep 9, 2024

v0.0.125

Sep 9, 2024

v0.0.124

Sep 9, 2024

v0.0.123

Sep 5, 2024

v0.0.122

Sep 5, 2024

v0.0.121

Sep 4, 2024

v0.0.120

Sep 2, 2024

v0.0.119

Sep 1, 2024

v0.0.118

Sep 1, 2024

Changes in this version

+ func IsKeyValueType[T any]() bool

+ func MakeKeyValueSerde[K comparable, V any](runtime StreamExecutionRuntime) serde.StreamKeyValueSerde[datastruct.KeyValue[K, V]]

+ func MakeMultiJoinLink[K comparable, T1, T2, R any](multiJoin TypedStream[R], stream TypedStream[datastruct.KeyValue[K, T2]])

+ func MakeSerde[T any](runtime StreamExecutionRuntime) serde.StreamSerde[T]

+ func MakeService[Runtime StreamExecutionRuntime, Cfg config.Config](name string, configSettings *config.ConfigSettings) Runtime

+ func RegisterKeyValueSerde[K comparable, V any](runtime StreamExecutionRuntime) serde.StreamKeyValueSerde[datastruct.KeyValue[K, V]]

+ func RegisterSerde[T any](runtime StreamExecutionRuntime) serde.StreamSerde[T]

+ type AppSinkStream struct

+ func MakeAppSinkStream[T any](name string, stream TypedStream[T], consumer ConsumerFunc[T]) *AppSinkStream[T]

+ func (s *AppSinkStream[T]) Consume(value T)

+ type BinaryConsumer interface

+ ConsumeBinary func([]byte)

+ type BinaryConsumerFunc func([]byte) error

+ func (f BinaryConsumerFunc) Consume(data []byte) error

+ type BinaryKVConsumer interface

+ ConsumeBinary func([]byte, []byte)

+ type BinaryKVConsumerFunc func([]byte, []byte) error

+ func (f BinaryKVConsumerFunc) Consume(key []byte, value []byte) error

+ type Caller interface

+ Consume func(value T)

+ type Collect interface

+ Out func(value T)

+ type ConsumeStatistics interface

+ Count func() int64

+ LinkId func() config.LinkId

+ type ConsumedStream struct

+ func (s *ConsumedStream[T]) Consume(value T)

+ func (s *ConsumedStream[T]) GetConsumer() TypedStreamConsumer[T]

+ func (s *ConsumedStream[T]) GetSerde() serde.StreamSerde[T]

+ func (s *ConsumedStream[T]) SetConsumer(consumer TypedStreamConsumer[T])

+ type Consumer interface

+ Consume func(T)

+ type ConsumerFunc func(T) error

+ func (f ConsumerFunc[T]) Consume(value T) error

+ type DataConnector interface

+ GetId func() int

+ GetName func() string

+ type DataSink interface

+ AddEndpoint func(SinkEndpoint)

+ GetDataConnector func() *config.DataConnectorConfig

+ GetEndpoint func(id int) SinkEndpoint

+ GetEndpoints func() []SinkEndpoint

+ GetRuntime func() StreamExecutionRuntime

+ Start func(context.Context) error

+ Stop func(context.Context)

+ type DataSinkEndpoint struct

+ func MakeDataSinkEndpoint(dataSink DataSink, config *config.EndpointConfig, ...) *DataSinkEndpoint

+ func (ep *DataSinkEndpoint) AddEndpointConsumer(endpointConsumer OutputEndpointConsumer)

+ func (ep *DataSinkEndpoint) GetConfig() *config.EndpointConfig

+ func (ep *DataSinkEndpoint) GetDataConnector() DataConnector

+ func (ep *DataSinkEndpoint) GetDataSink() DataSink

+ func (ep *DataSinkEndpoint) GetEndpointConsumers() []OutputEndpointConsumer

+ func (ep *DataSinkEndpoint) GetId() int

+ func (ep *DataSinkEndpoint) GetName() string

+ func (ep *DataSinkEndpoint) GetRuntime() StreamExecutionRuntime

+ type DataSinkEndpointConsumer struct

+ func MakeDataSinkEndpointConsumer[T any](endpoint SinkEndpoint, stream TypedSinkStream[T]) *DataSinkEndpointConsumer[T]

+ func (ec *DataSinkEndpointConsumer[T]) Endpoint() SinkEndpoint

+ func (ec *DataSinkEndpointConsumer[T]) GetEndpointWriter() TypedEndpointWriter[T]

+ type DataSource interface

+ AddEndpoint func(InputEndpoint)

+ GetDataConnector func() *config.DataConnectorConfig

+ GetEndpoint func(id int) InputEndpoint

+ GetEndpoints func() []InputEndpoint

+ GetRuntime func() StreamExecutionRuntime

+ Start func(context.Context) error

+ Stop func(context.Context)

+ type DataSourceEndpoint struct

+ func MakeDataSourceEndpoint(dataSource DataSource, config *config.EndpointConfig, ...) *DataSourceEndpoint

+ func (ep *DataSourceEndpoint) AddEndpointConsumer(endpointConsumer InputEndpointConsumer)

+ func (ep *DataSourceEndpoint) GetConfig() *config.EndpointConfig

+ func (ep *DataSourceEndpoint) GetDataConnector() DataConnector

+ func (ep *DataSourceEndpoint) GetDataSource() DataSource

+ func (ep *DataSourceEndpoint) GetEndpointConsumers() []InputEndpointConsumer

+ func (ep *DataSourceEndpoint) GetId() int

+ func (ep *DataSourceEndpoint) GetName() string

+ func (ep *DataSourceEndpoint) GetRuntime() StreamExecutionRuntime

+ type DataSourceEndpointConsumer struct

+ func MakeDataSourceEndpointConsumer[T any](endpoint InputEndpoint, inputStream TypedInputStream[T]) *DataSourceEndpointConsumer[T]

+ func (ec *DataSourceEndpointConsumer[T]) Consume(value T)

+ func (ec *DataSourceEndpointConsumer[T]) Endpoint() InputEndpoint

+ func (ec *DataSourceEndpointConsumer[T]) GetEndpointReader() TypedEndpointReader[T]

+ type DelayFunc func(T) error

+ type DelayFunction interface

+ Duration func(Stream, T) time.Duration

+ type DelayFunctionContext struct

+ type DelayStream struct

+ func MakeDelayStream[T any](name string, stream TypedStream[T], f DelayFunction[T]) *DelayStream[T]

+ func (s *DelayStream[T]) Consume(value T)

+ type Edge struct

+ Arrows string

+ Color struct{ ... }

+ From int

+ Label string

+ Length int

+ To int

+ type Endpoint interface

+ GetDataConnector func() DataConnector

+ GetId func() int

+ GetName func() string

+ type EndpointReader interface

+ type EndpointWriter interface

+ type FilterFunction interface

+ Filter func(Stream, T) bool

+ type FilterFunctionContext struct

+ type FilterStream struct

+ func MakeFilterStream[T any](name string, stream TypedStream[T], f FilterFunction[T]) *FilterStream[T]

+ func (s *FilterStream[T]) Consume(value T)

+ type FlatMapFunction interface

+ FlatMap func(Stream, T, Collect[R])

+ type FlatMapFunctionContext struct

+ type FlatMapIterableStream struct

+ func MakeFlatMapIterableStream[T, R any](name string, stream TypedStream[T]) *FlatMapIterableStream[T, R]

+ func (s *FlatMapIterableStream[T, R]) Consume(value T)

+ type FlatMapStream struct

+ func MakeFlatMapStream[T, R any](name string, stream TypedStream[T], f FlatMapFunction[T, R]) *FlatMapStream[T, R]

+ func (s *FlatMapStream[T, R]) Consume(value T)

+ type ForEachFunction interface

+ ForEach func(Stream, T)

+ type ForEachFunctionContext struct

+ type ForEachStream struct

+ func MakeForEachStream[T any](name string, stream TypedStream[T], f ForEachFunction[T]) *ForEachStream[T]

+ func (s *ForEachStream[T]) Consume(value T)

+ type InStubKVStream struct

+ func MakeInStubKVStream[T any](name string, runtime StreamExecutionRuntime) *InStubKVStream[T]

+ func (s *InStubKVStream[T]) ConsumeBinary(key []byte, value []byte)

+ type InStubStream struct

+ func MakeInStubStream[T any](name string, runtime StreamExecutionRuntime) *InStubStream[T]

+ func (s *InStubStream[T]) ConsumeBinary(data []byte)

+ type InputDataSource struct

+ func MakeInputDataSource(dataConnector *config.DataConnectorConfig, runtime StreamExecutionRuntime) *InputDataSource

+ func (ds *InputDataSource) AddEndpoint(endpoint InputEndpoint)

+ func (ds *InputDataSource) GetDataConnector() *config.DataConnectorConfig

+ func (ds *InputDataSource) GetEndpoint(id int) InputEndpoint

+ func (ds *InputDataSource) GetEndpoints() []InputEndpoint

+ func (ds *InputDataSource) GetId() int

+ func (ds *InputDataSource) GetName() string

+ func (ds *InputDataSource) GetRuntime() StreamExecutionRuntime

+ type InputEndpoint interface

+ AddEndpointConsumer func(consumer InputEndpointConsumer)

+ GetConfig func() *config.EndpointConfig

+ GetDataSource func() DataSource

+ GetEndpointConsumers func() []InputEndpointConsumer

+ GetRuntime func() StreamExecutionRuntime

+ type InputEndpointConsumer interface

+ Endpoint func() InputEndpoint

+ type InputKVSplitStream struct

+ func MakeInputKVSplitStream[T any](name string, runtime StreamExecutionRuntime) *InputKVSplitStream[T]

+ func (s *InputKVSplitStream[T]) ConsumeBinary(key []byte, value []byte)

+ type InputSplitStream struct

+ func MakeInputSplitStream[T any](name string, runtime StreamExecutionRuntime) *InputSplitStream[T]

+ func (s *InputSplitStream[T]) ConsumeBinary(data []byte)

+ type InputStream struct

+ func MakeInputStream[T any](name string, streamExecutionRuntime StreamExecutionRuntime) *InputStream[T]

+ func (s *InputStream[T]) GetEndpointId() int

+ type JoinFunction interface

+ Join func(Stream, K, []T1, []T2, Collect[R]) bool

+ type JoinFunctionContext struct

+ type JoinLink struct

+ func (s *JoinLink[K, T1, T2, R]) Consume(value datastruct.KeyValue[K, T2])

+ func (s *JoinLink[K, T1, T2, R]) GetConfig() *config.StreamConfig

+ func (s *JoinLink[K, T1, T2, R]) GetId() int

+ func (s *JoinLink[K, T1, T2, R]) GetName() string

+ func (s *JoinLink[K, T1, T2, R]) GetRuntime() StreamExecutionRuntime

+ func (s *JoinLink[K, T1, T2, R]) GetTransformationName() string

+ func (s *JoinLink[K, T1, T2, R]) GetTypeName() string

+ type JoinStream struct

+ func MakeJoinStream[K comparable, T1, T2, R any](name string, stream TypedStream[datastruct.KeyValue[K, T1]], ...) *JoinStream[K, T1, T2, R]

+ func (s *JoinStream[K, T1, T2, R]) Consume(value datastruct.KeyValue[K, T1])

+ func (s *JoinStream[K, T1, T2, R]) ConsumeRight(value datastruct.KeyValue[K, T2])

+ func (s *JoinStream[K, T1, T2, R]) Out(value R)

+ type KeyByFunction interface

+ KeyBy func(Stream, T) datastruct.KeyValue[K, V]

+ type KeyByFunctionContext struct

+ type KeyByStream struct

+ func MakeKeyByStream[T any, K comparable, V any](name string, stream TypedStream[T], f KeyByFunction[T, K, V]) *KeyByStream[T, K, V]

+ func (s *KeyByStream[T, K, V]) Consume(value T)

+ type LinkStream struct

+ func MakeLinkStream[T any](name string, runtime StreamExecutionRuntime) *LinkStream[T]

+ func (s *LinkStream[T]) Consume(value T)

+ func (s *LinkStream[T]) SetSource(stream TypedConsumedStream[T])

+ type MapFunction interface

+ Map func(Stream, T) R

+ type MapFunctionContext struct

+ type MapStream struct

+ func MakeMapStream[T, R any](name string, stream TypedStream[T], f MapFunction[T, R]) *MapStream[T, R]

+ func (s *MapStream[T, R]) Consume(value T)

+ type MergeLink struct

+ func (s *MergeLink[T]) Consume(value T)

+ func (s *MergeLink[T]) GetConfig() *config.StreamConfig

+ func (s *MergeLink[T]) GetId() int

+ func (s *MergeLink[T]) GetName() string

+ func (s *MergeLink[T]) GetRuntime() StreamExecutionRuntime

+ func (s *MergeLink[T]) GetTransformationName() string

+ func (s *MergeLink[T]) GetTypeName() string

+ type MergeStream struct

+ func MakeMergeStream[T any](name string, streams ...TypedStream[T]) *MergeStream[T]

+ type MultiJoinFunction interface

+ MultiJoin func(Stream, K, [][]interface{}, Collect[R]) bool

+ type MultiJoinFunctionContext struct

+ type MultiJoinLinkStream struct

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) Consume(value datastruct.KeyValue[K, T2])

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetConfig() *config.StreamConfig

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetId() int

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetName() string

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetRuntime() StreamExecutionRuntime

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetTransformationName() string

+ func (s *MultiJoinLinkStream[K, T1, T2, R]) GetTypeName() string

+ type MultiJoinStream struct

+ func MakeMultiJoinStream[K comparable, T, R any](name string, leftStream TypedStream[datastruct.KeyValue[K, T]], ...) *MultiJoinStream[K, T, R]

+ func (s *MultiJoinStream[K, T, R]) Consume(value datastruct.KeyValue[K, T])

+ func (s *MultiJoinStream[K, T, R]) ConsumeRight(index int, value datastruct.KeyValue[K, interface{}])

+ func (s *MultiJoinStream[K, T, R]) Out(value R)

+ type NetworkData struct

+ Edges []*Edge

+ Nodes []*Node

+ type Node struct

+ Color struct{ ... }

+ Id int

+ Label string

+ Opacity float32

+ X int

+ Y int

+ type OutStubBinaryKVStream struct

+ func MakeOutStubBinaryKVStream[T any](name string, stream TypedStream[T], consumer BinaryKVConsumerFunc) *OutStubBinaryKVStream[T]

+ func (s *OutStubBinaryKVStream[T]) Consume(value T)

+ type OutStubBinaryStream struct

+ func MakeOutStubBinaryStream[T any](name string, stream TypedStream[T], consumer BinaryConsumerFunc) *OutStubBinaryStream[T]

+ func (s *OutStubBinaryStream[T]) Consume(value T)

+ type OutStubStream struct

+ func MakeOutStubStream[T any](name string, stream TypedStream[T], consumer ConsumerFunc[T]) *OutStubStream[T]

+ func (s *OutStubStream[T]) Consume(value T)

+ type OutputDataSink struct

+ func MakeOutputDataSink(dataConnector *config.DataConnectorConfig, runtime StreamExecutionRuntime) *OutputDataSink

+ func (ds *OutputDataSink) AddEndpoint(endpoint SinkEndpoint)

+ func (ds *OutputDataSink) GetDataConnector() *config.DataConnectorConfig

+ func (ds *OutputDataSink) GetEndpoint(id int) SinkEndpoint

+ func (ds *OutputDataSink) GetEndpoints() []SinkEndpoint

+ func (ds *OutputDataSink) GetId() int

+ func (ds *OutputDataSink) GetName() string

+ func (ds *OutputDataSink) GetRuntime() StreamExecutionRuntime

+ type OutputEndpointConsumer interface

+ Endpoint func() SinkEndpoint

+ type ParallelsFunction interface

+ Parallels func(Stream, T, Collect[R])

+ type ParallelsFunctionContext struct

+ type ParallelsStream struct

+ func MakeParallelsStream[T, R any](name string, stream TypedStream[T], f ParallelsFunction[T, R]) *ParallelsStream[T, R]

+ func (s *ParallelsStream[T, R]) Consume(value T)

+ type ServiceApp struct

+ func (app *ServiceApp) AddDataSink(dataSink DataSink)

+ func (app *ServiceApp) AddDataSource(dataSource DataSource)

+ func (app *ServiceApp) Delay(duration time.Duration, f func())

+ func (app *ServiceApp) GetConfig() *config.ServiceAppConfig

+ func (app *ServiceApp) GetConsumeTimeout(from int, to int) time.Duration

+ func (app *ServiceApp) GetDataSink(id int) DataSink

+ func (app *ServiceApp) GetDataSource(id int) DataSource

+ func (app *ServiceApp) GetEndpointReader(endpoint Endpoint, stream Stream, valueType reflect.Type) EndpointReader

+ func (app *ServiceApp) GetEndpointWriter(endpoint Endpoint, stream Stream, valueType reflect.Type) EndpointWriter

+ func (app *ServiceApp) GetMetrics() metrics.Metrics

+ func (app *ServiceApp) GetServiceConfig() *config.ServiceConfig

+ func (app *ServiceApp) ReloadConfig(config config.Config)

+ func (app *ServiceApp) Start(ctx context.Context) error

+ func (app *ServiceApp) Stop(ctx context.Context)

+ type ServiceStream interface

+ type SinkEndpoint interface

+ AddEndpointConsumer func(consumer OutputEndpointConsumer)

+ GetConfig func() *config.EndpointConfig

+ GetDataSink func() DataSink

+ GetEndpointConsumers func() []OutputEndpointConsumer

+ GetRuntime func() StreamExecutionRuntime

+ type SinkStream struct

+ func MakeSinkStream[T any](name string, stream TypedStream[T]) *SinkStream[T]

+ func (s *SinkStream[T]) Consume(value T)

+ func (s *SinkStream[T]) GetEndpointId() int

+ func (s *SinkStream[T]) SetConsumer(consumer Consumer[T])

+ type SplitLink struct

+ func (s *SplitLink[T]) Consume(value T)

+ func (s *SplitLink[T]) GetConfig() *config.StreamConfig

+ func (s *SplitLink[T]) GetConsumer() TypedStreamConsumer[T]

+ func (s *SplitLink[T]) GetId() int

+ func (s *SplitLink[T]) GetName() string

+ func (s *SplitLink[T]) GetRuntime() StreamExecutionRuntime

+ func (s *SplitLink[T]) GetSerde() serde.StreamSerde[T]

+ func (s *SplitLink[T]) GetTransformationName() string

+ func (s *SplitLink[T]) GetTypeName() string

+ func (s *SplitLink[T]) SetConsumer(consumer TypedStreamConsumer[T])

+ type SplitStream struct

+ func MakeSplitStream[T any](name string, stream TypedStream[T]) *SplitStream[T]

+ func (s *SplitStream[T]) AddStream() TypedConsumedStream[T]

+ func (s *SplitStream[T]) Consume(value T)

+ type Stream interface

+ GetConfig func() *config.StreamConfig

+ GetId func() int

+ GetName func() string

+ GetRuntime func() StreamExecutionRuntime

+ GetTransformationName func() string

+ GetTypeName func() string

+ type StreamBase struct

+ func (s *StreamBase[T]) GetConfig() *config.StreamConfig

+ func (s *StreamBase[T]) GetId() int

+ func (s *StreamBase[T]) GetName() string

+ func (s *StreamBase[T]) GetRuntime() StreamExecutionRuntime

+ func (s *StreamBase[T]) GetTransformationName() string

+ func (s *StreamBase[T]) GetTypeName() string

+ type StreamExecutionEnvironment interface

+ AddDataSink func(dataSink DataSink)

+ AddDataSource func(dataSource DataSource)

+ Delay func(duration time.Duration, f func())

+ GetConfig func() *config.ServiceAppConfig

+ GetConsumeTimeout func(from int, to int) time.Duration

+ GetDataSink func(id int) DataSink

+ GetDataSource func(id int) DataSource

+ GetEndpointReader func(endpoint Endpoint, stream Stream, valueType reflect.Type) EndpointReader

+ GetEndpointWriter func(endpoint Endpoint, stream Stream, valueType reflect.Type) EndpointWriter

+ GetMetrics func() metrics.Metrics

+ GetSerde func(valueType reflect.Type) (serde.Serializer, error)

+ GetServiceConfig func() *config.ServiceConfig

+ SetConfig func(config config.Config)

+ Start func(context.Context) error

+ Stop func(context.Context)

+ StreamsInit func(ctx context.Context)

+ type StreamExecutionRuntime interface

+ type StreamFunction struct

+ func (f *StreamFunction[T]) AfterCall()

+ func (f *StreamFunction[T]) BeforeCall()

+ type TypedBinaryConsumedStream interface

+ type TypedBinaryKVConsumedStream interface

+ type TypedBinaryKVSplitStream interface

+ AddStream func() TypedConsumedStream[T]

+ type TypedBinarySplitStream interface

+ AddStream func() TypedConsumedStream[T]

+ type TypedConsumedStream interface

+ type TypedEndpointReader interface

+ Read func(io.Reader) (T, error)

+ type TypedEndpointWriter interface

+ Write func(T, io.Writer) error

+ type TypedInputStream interface

+ GetEndpointId func() int

+ type TypedJoinConsumedStream interface

+ ConsumeRight func(datastruct.KeyValue[K, T2])

+ type TypedLinkStream interface

+ SetSource func(TypedConsumedStream[T])

+ type TypedMultiJoinConsumedStream interface

+ ConsumeRight func(int, datastruct.KeyValue[K, interface{}])

+ type TypedSinkStream interface

+ GetEndpointId func() int

+ SetConsumer func(Consumer[T])

+ type TypedSplitStream interface

+ AddStream func() TypedConsumedStream[T]

+ type TypedStream interface

+ GetConsumer func() TypedStreamConsumer[T]

+ GetSerde func() serde.StreamSerde[T]

+ SetConsumer func(TypedStreamConsumer[T])

+ type TypedStreamConsumer interface

+ type TypedTransformConsumedStream interface

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL