Package org.nuxeo.lib.stream.computation.log

Class LogStreamManager

java.lang.Object

org.nuxeo.lib.stream.computation.log.LogStreamManager

All Implemented Interfaces:

AutoCloseable, StreamManager

public class LogStreamManager
extends Object
implements StreamManager

StreamManager based on a LogManager

Since:

11.1

Field Summary

Fields

Modifier and Type	Field	Description
protected final Map<Name,RecordFilterChain>	filters
static final Codec<Record>	INTERNAL_CODEC
protected final LogManager	logManager
static final String	METRICS_STREAM
protected final Map<String,StreamProcessor>	processors
static final String	PROCESSORS_STREAM
protected final Map<String,Settings>	settings
protected final Set<Name>	streams
protected final Map<String,String>	systemMetadata
protected final Map<String,Topology>	topologies

Constructor Summary

Constructors

Constructor	Description
LogStreamManager(String nodeId, LogManager logManager)
LogStreamManager(LogManager logManager)

Method Summary

Modifier and Type	Method	Description
LogOffset	append(String streamUrn, Record record)	Appends a record to a processor's source stream.
void	close()
StreamProcessor	createStreamProcessor(String processorName)	Creates a registered processor without starting it.
LogTailer<Record>	createTailer(Name computationName, Collection<LogPartition> streamPartitions)
protected Codec<Record>	getCodec(Collection<Name> streams)
RecordFilter	getFilter(Name stream)
LogManager	getLogManager()
protected String	getNodeId()
StreamProcessor	getProcessor(String processorName)	Gets a processor.
Set<String>	getProcessorNames()	Gets a set of processor names.
protected void	initAppenders(Collection<String> streams, Settings settings)
protected void	initInternalStream(Name stream)
protected void	initInternalStreams()
protected void	initStream(String streamName, Settings settings)
protected void	initStreams(Topology topology, Settings settings)
void	register(String processorName, Topology topology, Settings settings)	Registers a processor and initializes the underlying streams, this is needed before creating a processor or appending record in source streams.
void	register(List<String> streams, Settings settings)	Registers some source Streams without any processors.
protected void	registerFilters(Collection<String> streams, Settings settings)
LogTailer<Record>	subscribe(Name computationName, Collection<Name> streams, RebalanceListener listener)
boolean	supportSubscribe()	Deprecated. since 2021.34 use supportSubscribe(Name) instead
boolean	supportSubscribe(Name stream)	Returns true if the subscribe(org.nuxeo.lib.stream.log.Name, java.util.Collection<org.nuxeo.lib.stream.log.Name>, org.nuxeo.lib.stream.log.RebalanceListener) method is supported for the specific stream.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.nuxeo.lib.stream.computation.StreamManager

registerAndCreateProcessor

Field Details

PROCESSORS_STREAM

public static final String PROCESSORS_STREAM

See Also:

• Constant Field Values

METRICS_STREAM

public static final String METRICS_STREAM

See Also:

• Constant Field Values

INTERNAL_CODEC

public static final Codec<Record> INTERNAL_CODEC

logManager

protected final LogManager logManager

systemMetadata

protected final Map<String,String> systemMetadata

processors

protected final Map<String,StreamProcessor> processors

topologies

protected final Map<String,Topology> topologies

settings

protected final Map<String,Settings> settings

filters

protected final Map<Name,RecordFilterChain> filters

streams

protected final Set<Name> streams

Constructor Details

LogStreamManager

public LogStreamManager(LogManager logManager)

LogStreamManager

public LogStreamManager(String nodeId,
 LogManager logManager)

Method Details

initInternalStreams

protected void initInternalStreams()

initInternalStream

protected void initInternalStream(Name stream)

register

public void register(String processorName,
 Topology topology,
 Settings settings)

Description copied from interface: StreamManager

Registers a processor and initializes the underlying streams, this is needed before creating a processor or appending record in source streams.

Specified by:

register in interface StreamManager

register

public void register(List<String> streams,
 Settings settings)

Description copied from interface: StreamManager

Registers some source Streams without any processors.

Specified by:

register in interface StreamManager

createStreamProcessor

public StreamProcessor createStreamProcessor(String processorName)

Description copied from interface: StreamManager

Creates a registered processor without starting it.

Specified by:

createStreamProcessor in interface StreamManager

getNodeId

protected String getNodeId()

getLogManager

public LogManager getLogManager()

append

public LogOffset append(String streamUrn,
 Record record)

Description copied from interface: StreamManager

Appends a record to a processor's source stream.

Specified by:

append in interface StreamManager

getProcessorNames

public Set<String> getProcessorNames()

Description copied from interface: StreamManager

Gets a set of processor names.

Specified by:

getProcessorNames in interface StreamManager

getProcessor

public StreamProcessor getProcessor(String processorName)

Description copied from interface: StreamManager

Gets a processor.

Specified by:

getProcessor in interface StreamManager

Returns:

null if the processor doesn't exist

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface StreamManager

supportSubscribe

public boolean supportSubscribe(Name stream)

Returns true if the subscribe(org.nuxeo.lib.stream.log.Name, java.util.Collection<org.nuxeo.lib.stream.log.Name>, org.nuxeo.lib.stream.log.RebalanceListener) method is supported for the specific stream.

Since:

2021.34

supportSubscribe

public boolean supportSubscribe()

Deprecated.

since 2021.34 use supportSubscribe(Name) instead

Returns true if the subscribe(org.nuxeo.lib.stream.log.Name, java.util.Collection<org.nuxeo.lib.stream.log.Name>, org.nuxeo.lib.stream.log.RebalanceListener) method is supported. Now deprecated because some implementations support subscribe only on specific streams.

subscribe

public LogTailer<Record> subscribe(Name computationName,
 Collection<Name> streams,
 RebalanceListener listener)

createTailer

public LogTailer<Record> createTailer(Name computationName,
 Collection<LogPartition> streamPartitions)

getFilter

public RecordFilter getFilter(Name stream)

getCodec

protected Codec<Record> getCodec(Collection<Name> streams)

initStreams

protected void initStreams(Topology topology,
 Settings settings)

initStream

protected void initStream(String streamName,
 Settings settings)

initAppenders

protected void initAppenders(Collection<String> streams,
 Settings settings)

registerFilters

protected void registerFilters(Collection<String> streams,
 Settings settings)