An implementation of a SampleInput. More...

#include <SampleInputStream.h>

Inheritance diagram for nidas::dynld::SampleInputStream:

Classes
struct	ReadResult
	Tuple for all the possible results of iostream reads. More...

Public Types
typedef nidas::core::dsm_time_t	dsm_time_t

Public Member Functions
	SampleInputStream (bool raw=false)
	Constructor. More...

	SampleInputStream (nidas::core::IOChannel *iochannel, bool raw=false)
	Constructor. More...

virtual SampleInputStream *	clone (nidas::core::IOChannel *iochannel)
	Create a clone, with a new, connected IOChannel. More...

virtual	~SampleInputStream ()

std::string	getName () const

int	getFd () const

virtual void	setIOChannel (nidas::core::IOChannel *val)
	Set the IOChannel for this SampleInputStream.h. More...

void	readInputHeader () throw (nidas::util::IOException)
	Read archive information at beginning of input stream or file. More...

bool	parseInputHeader () throw (nidas::util::IOException)

const nidas::core::SampleInputHeader &	getInputHeader () const

void	requestConnection (nidas::core::DSMService *) throw (nidas::util::IOException)

virtual nidas::core::SampleInput *	getOriginal () const

nidas::core::SampleInput *	connected (nidas::core::IOChannel *iochan) throw ()
	Implementation of IOChannelRequester::connected. More...

void	setNonBlocking (bool val) throw (nidas::util::IOException)

bool	isNonBlocking () const throw (nidas::util::IOException)

const nidas::core::DSMConfig *	getDSMConfig () const
	What DSM am I connnected to? May be NULL if it cannot be determined. More...

void	setKeepStats (bool val)

void	addSampleTag (const nidas::core::SampleTag *tag) throw ()
	Implementation of SampleInput::addSampleTag(). More...

void	removeSampleTag (const nidas::core::SampleTag *tag) throw ()

nidas::core::SampleSource *	getRawSampleSource ()
	Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples. More...

nidas::core::SampleSource *	getProcessedSampleSource ()
	Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples. More...

std::list< const nidas::core::SampleTag * >	getSampleTags () const
	Get the output SampleTags. More...

nidas::core::SampleTagIterator	getSampleTagIterator () const
	Implementation of SampleSource::getSampleTagIterator(). More...

void	addSampleClient (nidas::core::SampleClient *client) throw ()
	Implementation of SampleSource::addSampleClient(). More...

void	removeSampleClient (nidas::core::SampleClient *client) throw ()
	Remove a SampleClient from this SampleSource. More...

void	addSampleClientForTag (nidas::core::SampleClient client, const nidas::core::SampleTag tag) throw ()
	Add a Client for a given SampleTag. More...

void	removeSampleClientForTag (nidas::core::SampleClient client, const nidas::core::SampleTag tag) throw ()
	Remove a SampleClient for a given SampleTag from this SampleSource. More...

int	getClientCount () const throw ()
	How many SampleClients are currently in my list. More...

void	flush () throw ()
	Implementation of SampleSource::flush(), unpacks and distributes any samples currently in the read buffer. More...

const nidas::core::SampleStats &	getSampleStats () const

bool	readSamples () throw (nidas::util::IOException)
	Read a buffer of data, serialize the data into samples, and distribute() samples to the receive() method of my SampleClients and DSMSensors. More...

void	search (const nidas::util::UTime &tt) throw (nidas::util::IOException)
	Search forward until a sample header is read whose time is greater than or equal to tt. More...

nidas::core::Sample *	readSample () throw (nidas::util::IOException)
	Read the next sample from the InputStream. More...

void	distribute (const nidas::core::Sample *s) throw ()
	Distribute a sample to my clients. More...

size_t	getBadSamples () const

void	close () throw (nidas::util::IOException)

void	setBadSampleFilter (const nidas::core::BadSampleFilter &bsf)
	Replace the bad sample filter rules for this stream with `bsf`. More...

void	setFilterBadSamples (bool val)
	See BadSampleFilter. More...

void	setMinDsmId (int val)
	See BadSampleFilter. More...

void	setMaxDsmId (int val)
	See BadSampleFilter. More...

void	setMinSampleLength (unsigned int val)
	See BadSampleFilter. More...

void	setMaxSampleLength (unsigned int val)
	See BadSampleFilter. More...

void	setMinSampleTime (const nidas::util::UTime &val)
	See BadSampleFilter. More...

void	setMaxSampleTime (const nidas::util::UTime &val)
	See BadSampleFilter. More...

void	fromDOMElement (const xercesc::DOMElement *node) throw (nidas::util::InvalidParameterException)
	Initialize myself from a xercesc::DOMElement. More...

void	setExpectHeader (bool val)

bool	getExpectHeader () const

virtual xercesc::DOMElement *	toDOMParent (xercesc::DOMElement *parent, bool complete) const throw (xercesc::DOMException)
	Create a DOMElement and append it to the parent. More...

virtual xercesc::DOMElement *	toDOMElement (xercesc::DOMElement *node, bool complete) const throw (xercesc::DOMException)
	Add my content into a DOMElement. More...

Static Public Member Functions
static const XMLCh *	getNamespaceURI ()

Protected Member Functions
	SampleInputStream (SampleInputStream &x, nidas::core::IOChannel *iochannel)
	Copy constructor, with a new, connected IOChannel. More...

Protected Attributes
nidas::core::IOChannel *	_iochan

nidas::core::SampleSourceSupport	_source

Private Member Functions
nidas::core::Sample *	nextSample () throw ()
	Unpack the next sample from the InputStream. More...

nidas::core::Sample *	nextSample (bool keepreading, bool searching=false, dsm_time_t search_time=LONG_LONG_MIN) throw (nidas::util::IOException)
	Unpack the next sample from the InputStream buffer or by reading more data if `keepreading` is true. More...

void	checkUnexpectedEOF ()

ReadResult	readBlock (bool keepreading, char *&ptr, size_t &lentoread)
	Read a block into memory, updating the given block pointer and length counter accordingly. More...

ReadResult	read (bool keepreading, char *ptr, size_t lentoread)
	All reads of the iostream go through here. More...

void	handleNewInput ()
	Whenever we are at the start of a new input, we need to start over with parsing the input header. More...

nidas::core::Sample *	handleEOF (bool keepreading)
	When EOF has been caught, then either we need to return the last pending sample, or else we need to throw the exception. More...

void	closeBlocks ()

nidas::core::Sample *	sampleFromHeader () throw ()
	Check the current header for validity and generate a sample for it. More...

	SampleInputStream (const SampleInputStream &x)
	No regular copy. More...

SampleInputStream &	operator= (const SampleInputStream &x)
	No assignment. More...

Private Attributes
nidas::core::DSMService *	_service
	Service that has requested my input. More...

nidas::core::IOStream *	_iostream

const nidas::core::DSMConfig *	_dsm

bool	_expectHeader

bool	_inputHeaderParsed

nidas::core::SampleHeader	_sheader

size_t	_headerToRead

char *	_hptr

nidas::core::Sample *	_samp
	Will be non-null if we have previously read part of a sample from the stream. More...

nidas::core::Sample *	_sampPending
	The currently pending sample. More...

size_t	_dataToRead
	How many bytes left to read from the stream into the data portion of samp. More...

char *	_dptr
	Pointer into the data portion of samp where we will read next. More...

bool	_skipSample
	This is set if the data for the current sample header should be read but skipped, because the sample is the first after a block of bad samples and thus the time at the front of the sample might still be corrupt. More...

BlockStats	_block
	Information about the current block of samples, good or bad. More...

size_t	_badSamples
	Number of bad samples in the stream so far, which is to say number of bytes checked which did not contain a reasonable sample header. More...

size_t	_goodSamples
	Number of good samples in the stream so far. More...

nidas::core::SampleInputHeader	_inputHeader

nidas::core::BadSampleFilter	_bsf

SampleInputStream *	_original

bool	_raw

std::string	_last_name
	Keep the current name of the input stream so it can be referenced even after the input stream has advanced to a new input name. More...

nidas::util::EOFException	_eofx

bool	_ateof

Detailed Description

An implementation of a SampleInput.

The readSamples method converts raw bytes from the iochannel into Samples.

If a SampleClient has requested processed Samples via addProcessedSampleClient, then SampleInputStream will pass Samples to the respective DSMSensor for processing, and then the DSMSensor passes the processed Samples to the SampleClient:

iochannel -> readSamples method -> DSMSensor -> SampleClient

If a SampleClient has requested non-processed Samples, via the simple addSampleClient method, then SampleInput stream passes the Samples straight to the SampleClient:

iochannel -> readSamples method -> SampleClient

Member Typedef Documentation

typedef nidas::core::dsm_time_t nidas::dynld::SampleInputStream::dsm_time_t

Constructor & Destructor Documentation

SampleInputStream::SampleInputStream ( bool raw = false )

Constructor.

Parameters

raw	Whether the input samples are raw.

Referenced by clone().

SampleInputStream::SampleInputStream	(	nidas::core::IOChannel *	iochannel,
		bool	raw = `false`
	)

Constructor.

Parameters

iochannel The IOChannel that we use for data input. SampleInputStream will own the pointer to the IOChannel, and will delete it in ~SampleInputStream(). If it is a null pointer, then it must be set within the fromDOMElement method.

References _iochan, _iostream, nidas::core::IOChannel::getBufferSize(), and setIOChannel().

SampleInputStream::~SampleInputStream ( )

virtual

References _iochan, _samp, _sampPending, close(), and nidas::core::Sample::freeReference().

SampleInputStream::SampleInputStream	(	SampleInputStream &	x,
		nidas::core::IOChannel *	iochannel
	)

protected

Copy constructor, with a new, connected IOChannel.

References _iochan, _iostream, nidas::core::IOChannel::getBufferSize(), and setIOChannel().

nidas::dynld::SampleInputStream::SampleInputStream ( const SampleInputStream & x )

private

No regular copy.

Member Function Documentation

void nidas::dynld::SampleInputStream::addSampleClient	(	nidas::core::SampleClient *	client	)
throw	(
	)

inlinevirtual

Implementation of SampleSource::addSampleClient().

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::addSampleClient().

void nidas::dynld::SampleInputStream::addSampleClientForTag	(	nidas::core::SampleClient *	client,
		const nidas::core::SampleTag *	tag
	)
throw	(
	)

inlinevirtual

Add a Client for a given SampleTag.

Implementation of SampleSource::addSampleClient().

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::addSampleClientForTag().

void nidas::dynld::SampleInputStream::addSampleTag	(	const nidas::core::SampleTag *	tag	)
throw	(
	)

inlinevirtual

Implementation of SampleInput::addSampleTag().

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::addSampleTag().

Referenced by nidas::dynld::raf::SyncServer::initSensors(), StatsProcess::run(), and DataPrep::run().

void SampleInputStream::checkUnexpectedEOF ( )

private

References _dataToRead, _headerToRead, _last_name, _samp, _sheader, CNAME, nidas::core::Sample::getDataByteLength(), nidas::core::SampleHeader::getSizeOf(), and WLOG.

Referenced by read().

SampleInputStream * SampleInputStream::clone ( nidas::core::IOChannel * iochannel )

virtual

Create a clone, with a new, connected IOChannel.

Reimplemented in nidas::dynld::isff::PacketInputStream, and nidas::dynld::RawSampleInputStream.

References SampleInputStream().

void SampleInputStream::close	(		)
throw	(	nidas::util::IOException
	)

virtual

Implements nidas::core::SampleInput.

References _iochan, _iostream, nidas::core::IOChannel::close(), and closeBlocks().

Referenced by NidsMerge::run(), StatsProcess::run(), DataPrep::run(), nidas::dynld::raf::SyncServer::stop(), and ~SampleInputStream().

void SampleInputStream::closeBlocks ( )

private

References _badSamples, _block, _bsf, _goodSamples, _last_name, nidas::dynld::BlockStats::block_start, CNAME, DLOG, nidas::core::BadSampleFilter::filterBadSamples(), nidas::util::LogContext::log(), LOG_DEBUG, LOG_WARNING, and nidas::dynld::BlockStats::nbytes.

Referenced by close(), and handleNewInput().

SampleInput * SampleInputStream::connected	(	nidas::core::IOChannel *	iochan	)
throw	(
	)

virtual

Implementation of IOChannelRequester::connected.

One can use this method to notify SampleInputStream that the IOChannel is connected, which will cause SampleInputStream to open the IOStream.

Implements nidas::core::IOChannelRequester.

void nidas::dynld::SampleInputStream::distribute	(	const nidas::core::Sample *	s	)
throw	(
	)

inline

Distribute a sample to my clients.

One could use this to insert a sample into the stream.

References _source, and nidas::core::SampleSourceSupport::distribute().

Referenced by nidas::dynld::raf::SyncServer::handleSample().

void SampleInputStream::flush	(	)
throw	(
	)

virtual

Implementation of SampleSource::flush(), unpacks and distributes any samples currently in the read buffer.

Implements nidas::core::SampleSource.

References _source, CNAME, nidas::core::SampleSourceSupport::distribute(), nidas::core::SampleSourceSupport::getClientCount(), nextSample(), and VLOG.

void SampleInputStream::fromDOMElement	(	const xercesc::DOMElement *	)
throw	(	nidas::util::InvalidParameterException
	)

virtual

Initialize myself from a xercesc::DOMElement.

Implements nidas::core::DOMable.

References nidas::core::IOChannel::createIOChannel(), nidas::core::DOMable::fromDOMElement(), nidas::core::XDOMAttr::getName(), and WLOG.

Referenced by nidas::dynld::RawSampleInputStream::fromDOMElement().

size_t nidas::dynld::SampleInputStream::getBadSamples ( ) const

inline

References _badSamples.

int nidas::dynld::SampleInputStream::getClientCount	(	)	const
throw	(
	)

inlinevirtual

How many SampleClients are currently in my list.

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::getClientCount().

const DSMConfig * SampleInputStream::getDSMConfig ( ) const

virtual

What DSM am I connnected to? May be NULL if it cannot be determined.

Implements nidas::core::SampleInput.

References _dsm.

bool nidas::dynld::SampleInputStream::getExpectHeader ( ) const

inline

References _expectHeader.

int SampleInputStream::getFd ( ) const

virtual

Implements nidas::core::SampleInput.

References _iochan, and nidas::core::IOChannel::getFd().

const nidas::core::SampleInputHeader& nidas::dynld::SampleInputStream::getInputHeader ( ) const

inline

References _inputHeader.

Referenced by nidas::dynld::raf::SyncRecordReader::init(), nidas::dynld::raf::SyncServer::openStream(), SensorExtract::run(), StatsProcess::run(), ExtractFast2D::run(), ExtractDMT::run(), and DataPrep::run().

string SampleInputStream::getName ( ) const

virtual

Implements nidas::core::SampleInput.

References _iochan, and nidas::core::IOChannel::getName().

Referenced by parseInputHeader(), readInputHeader(), MergeVerifier::reportBackward(), MergeVerifier::reportDuplicate(), MergeVerifier::reportMissing(), MergeVerifier::run(), and sampleFromHeader().

static const XMLCh* nidas::core::DOMable::getNamespaceURI ( )

inlinestaticinherited

References nidas::core::DOMable::namespaceURI.

Referenced by nidas::dynld::UDPSampleOutput::getProjectDOM(), nidas::core::Dataset::toDOMElement(), nidas::core::ProjectConfig::toDOMElement(), nidas::core::Dataset::toDOMParent(), nidas::core::ProjectConfig::toDOMParent(), nidas::core::Datasets::toDOMParent(), nidas::core::ProjectConfigs::toDOMParent(), nidas::core::DSMConfig::toDOMParent(), nidas::core::Project::toDOMParent(), nidas::core::Site::toDOMParent(), nidas::core::SampleTag::toDOMParent(), nidas::core::ServerSocket::toDOMParent(), nidas::core::Variable::toDOMParent(), nidas::core::DSMSensor::toDOMParent(), nidas::core::Datasets::writeXML(), and nidas::core::ProjectConfigs::writeXML().

virtual nidas::core::SampleInput* nidas::dynld::SampleInputStream::getOriginal ( ) const

inlinevirtual

Implements nidas::core::SampleInput.

References _original.

nidas::core::SampleSource* nidas::dynld::SampleInputStream::getProcessedSampleSource ( )

inlinevirtual

Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples.

SampleClients use this method to get a pointer to whatever sample source they are interested in. Derived classes can return NULL if they are not a SampleSource of processed samples.

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::getProcessedSampleSource().

nidas::core::SampleSource* nidas::dynld::SampleInputStream::getRawSampleSource ( )

inlinevirtual

Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples.

SampleClients use this method to get a pointer to whatever sample source they are interested in. Derived classes can return NULL if they are not a SampleSource of raw samples.

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::getRawSampleSource().

const nidas::core::SampleStats& nidas::dynld::SampleInputStream::getSampleStats ( ) const

inlinevirtual

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::getSampleStats().

nidas::core::SampleTagIterator nidas::dynld::SampleInputStream::getSampleTagIterator ( ) const

inlinevirtual

Implementation of SampleSource::getSampleTagIterator().

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::getSampleTagIterator().

std::list<const nidas::core::SampleTag*> nidas::dynld::SampleInputStream::getSampleTags ( ) const

inlinevirtual

Get the output SampleTags.

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::getSampleTags().

Sample * SampleInputStream::handleEOF ( bool keepreading )

private

When EOF has been caught, then either we need to return the last pending sample, or else we need to throw the exception.

HOWEVER, because nextSample() is not allowed to throw an IOException when keepreading is false, since in general it does not make sense to trigger EOF when nothing is being read, this handler only throws the exception when keepreading is true.

References _eofx, _sampPending, CNAME, and DLOG.

Referenced by readSamples().

void SampleInputStream::handleNewInput ( )

private

Whenever we are at the start of a new input, we need to start over with parsing the input header.

This method sets up the right state to start over on a new file.

References _inputHeaderParsed, _iostream, nidas::core::IOStream::backup(), closeBlocks(), CNAME, DLOG, nidas::core::IOStream::getNumInputBytes(), and VLOG.

Referenced by read().

bool SampleInputStream::isNonBlocking	(		)	const
throw	(	nidas::util::IOException
	)

virtual

Implements nidas::core::SampleInput.

References _iochan, and nidas::core::IOChannel::isNonBlocking().

Sample * SampleInputStream::nextSample	(	)
throw	(
	)

private

Unpack the next sample from the InputStream.

This method does not perform any physical reads.

Returns: pointer to a sample if there is one available in the buffer, else NULL.

Referenced by flush(), readSample(), and readSamples().

nidas::core::Sample* nidas::dynld::SampleInputStream::nextSample	(	bool	keepreading,
		bool	searching = `false`,
		dsm_time_t	search_time = `LONG_LONG_MIN`
	)
throw	(	nidas::util::IOException
	)

private

Unpack the next sample from the InputStream buffer or by reading more data if keepreading is true.

If searching is set, then reading stops after the first sample header found whose time tag is after search_time, as described in the search() method.

SampleInputStream& nidas::dynld::SampleInputStream::operator= ( const SampleInputStream & x )

private

No assignment.

bool SampleInputStream::parseInputHeader	(		)
throw	(	nidas::util::IOException
	)

SampleInputStream::ReadResult SampleInputStream::read	(	bool	keepreading,
		char *	ptr,
		size_t	lentoread
	)

private

All reads of the iostream go through here.

There are three kinds of reads needed:

Read to fill the iostream buffer but not a local block. (ptr==0 and keepreading=true)
Read from the iostream buffer into a local block (keepreading==false).
Read from the iostream to fill a local block, filling the buffer as needed (keepreading=true)

In all these cases, we need to catch if we hit the end of a file to know if the file ended unexpectedly. Also, if we hit the end of a file and a sample is still pending (because filtering is enabled and a sample is not good until it passes the filter and is succeeded by eof or another good sample), then we also need to return that sample. This method records the eof exception but does not pass it on, waiting instead for it to be thrown after any pending sample has been returned by nextSample(). Callers can test ReadResult members for which cases have occurred.

References _ateof, _eofx, _iostream, nidas::core::IOStream::available(), checkUnexpectedEOF(), CNAME, DLOG, nidas::dynld::SampleInputStream::ReadResult::eof, handleNewInput(), nidas::core::IOStream::isNewInput(), nidas::dynld::SampleInputStream::ReadResult::len, nidas::dynld::SampleInputStream::ReadResult::newinput, nidas::core::IOStream::read(), nidas::core::IOStream::readBuf(), and VLOG.

Referenced by readBlock(), and readSamples().

SampleInputStream::ReadResult SampleInputStream::readBlock	(	bool	keepreading,
		char *&	ptr,
		size_t &	lentoread
	)

private

Read a block into memory, updating the given block pointer and length counter accordingly.

If keepreading is false, only read into the block what is available from the iostream buffer. Return the ReadResult of the last read. Always return if the current input ends or eof is reached.

References nidas::dynld::SampleInputStream::ReadResult::eof, nidas::dynld::SampleInputStream::ReadResult::len, nidas::dynld::SampleInputStream::ReadResult::newinput, and read().

void SampleInputStream::readInputHeader	(		)
throw	(	nidas::util::IOException
	)

Read archive information at beginning of input stream or file.

References _inputHeaderParsed, _iostream, _last_name, CNAME, DLOG, getName(), nidas::core::IOStream::getNumInputBytes(), parseInputHeader(), and nidas::core::IOStream::read().

Referenced by nidas::dynld::raf::SyncRecordReader::init(), nidas::dynld::raf::SyncServer::openStream(), DataStats::readHeader(), MergeVerifier::run(), SensorExtract::run(), StatsProcess::run(), ExtractDMT::run(), ExtractFast2D::run(), and DataPrep::run().

Sample * SampleInputStream::readSample	(		)
throw	(	nidas::util::IOException
	)

virtual

Read the next sample from the InputStream.

The caller must call freeReference on the sample when they're done with it. This method may perform zero or more reads of the IOChannel.

Returns: pointer to a sample, never NULL.

Implements nidas::core::SampleInput.

References nextSample().

Referenced by nidas::dynld::raf::SyncRecordReader::init(), nidas::dynld::raf::SyncServer::openStream(), nidas::dynld::raf::SyncServer::read(), NidsMerge::run(), MergeVerifier::run(), SensorExtract::run(), ExtractDMT::run(), and ExtractFast2D::run().

bool SampleInputStream::readSamples	(		)
throw	(	nidas::util::IOException
	)

virtual

Read a buffer of data, serialize the data into samples, and distribute() samples to the receive() method of my SampleClients and DSMSensors.

This will perform only one physical read of the underlying IOChannel and so is appropriate to use when a select() or poll() has determined that there is data available on the file descriptor, or when the physical device is configured for non-blocking reads.

Returns: false: no data available for physical read, likely the result of an EAGAIN from a non-blocking read on a file descriptor. true: physical read did not necessarily consume all available data.

Implements nidas::core::SampleInput.

References _ateof, _iostream, _source, nidas::core::IOStream::available(), CNAME, nidas::core::SampleSourceSupport::distribute(), nidas::dynld::SampleInputStream::ReadResult::eof, handleEOF(), nidas::dynld::SampleInputStream::ReadResult::len, nidas::dynld::SampleInputStream::ReadResult::newinput, nextSample(), read(), and VLOG.

Referenced by DataStats::readSamples(), StatsProcess::run(), and DataPrep::run().

void nidas::dynld::SampleInputStream::removeSampleClient	(	nidas::core::SampleClient *	c	)
throw	(
	)

inlinevirtual

Remove a SampleClient from this SampleSource.

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::removeSampleClient().

void nidas::dynld::SampleInputStream::removeSampleClientForTag	(	nidas::core::SampleClient *	c,
		const nidas::core::SampleTag *
	)
throw	(
	)

inlinevirtual

Remove a SampleClient for a given SampleTag from this SampleSource.

The pointer to the SampleClient must remain valid, until after it is removed.

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::removeSampleClientForTag().

void nidas::dynld::SampleInputStream::removeSampleTag	(	const nidas::core::SampleTag *	tag	)
throw	(
	)

inlinevirtual

Implements nidas::core::SampleSource.

References _source, and nidas::core::SampleSourceSupport::removeSampleTag().

void SampleInputStream::requestConnection	(	nidas::core::DSMService *	requester	)
throw	(	nidas::util::IOException
	)

virtual

Implements nidas::core::SampleInput.

Sample * SampleInputStream::sampleFromHeader	(	)
throw	(
	)

private

Check the current header for validity and generate a sample for it.

References _badSamples, _block, _bsf, _goodSamples, _headerToRead, _hptr, _iostream, _sheader, _skipSample, nidas::util::LogContext::active(), nidas::dynld::BlockStats::addGoodSample(), CNAME, nidas::dynld::BlockStats::endBadBlock(), nidas::core::BadSampleFilter::filterBadSamples(), ftime(), nidas::core::SampleHeader::getDataByteLength(), nidas::core::SampleHeader::getId(), getName(), nidas::core::IOStream::getNumInputBytes(), nidas::core::getSample(), nidas::core::SampleHeader::getSizeOf(), nidas::core::SampleHeader::getTimeTag(), nidas::core::SampleHeader::getType(), nidas::dynld::BlockStats::good, nidas::core::BadSampleFilter::invalidSampleHeader(), nidas::util::LogContext::log(), LOG_DEBUG, LOG_WARNING, nidas::dynld::BlockStats::nbytes, nidas::core::Sample::setId(), nidas::core::Sample::setTimeTag(), nidas::dynld::BlockStats::start_time, nidas::dynld::BlockStats::startBadBlock(), and WLOG.

void SampleInputStream::search	(	const nidas::util::UTime &	tt	)
throw	(	nidas::util::IOException
	)

Search forward until a sample header is read whose time is greater than or equal to tt.

Leaves the InputStream positioned so that the next call to readSample() or readSamples() will read the rest of the sample.

References CNAME, and DLOG.

Referenced by StatsProcess::run(), and DataPrep::run().

void nidas::dynld::SampleInputStream::setBadSampleFilter ( const nidas::core::BadSampleFilter & bsf )

inline

Replace the bad sample filter rules for this stream with bsf.

References _bsf.

Referenced by StatsProcess::run(), DataPrep::run(), and DataStats::run().

void nidas::dynld::SampleInputStream::setExpectHeader ( bool val )

inline

References _expectHeader.

Referenced by setIOChannel().

void nidas::dynld::SampleInputStream::setFilterBadSamples ( bool val )

inline

See BadSampleFilter.

References _bsf, and nidas::core::BadSampleFilter::setFilterBadSamples().

void SampleInputStream::setIOChannel ( nidas::core::IOChannel * val )

virtual

Set the IOChannel for this SampleInputStream.h.

References _dsm, _iochan, nidas::core::IOChannel::close(), nidas::core::Project::findDSM(), nidas::core::IOChannel::getConnectionInfo(), nidas::util::Inet4SocketAddress::getInet4Address(), nidas::core::Project::getInstance(), nidas::core::ConnectionInfo::getRemoteSocketAddress(), setExpectHeader(), and nidas::core::IOChannel::writeNidasHeader().

Referenced by SampleInputStream().

void nidas::dynld::SampleInputStream::setKeepStats ( bool val )

inlinevirtual

Implements nidas::core::SampleInput.

References _source, and nidas::core::SampleSourceSupport::setKeepStats().

void nidas::dynld::SampleInputStream::setMaxDsmId ( int val )

inline

See BadSampleFilter.

References _bsf, and nidas::core::BadSampleFilter::setMaxDsmId().

Referenced by nidas::dynld::raf::SyncServer::openStream().

void nidas::dynld::SampleInputStream::setMaxSampleLength ( unsigned int val )

inline

See BadSampleFilter.

References _bsf, and nidas::core::BadSampleFilter::setMaxSampleLength().

Referenced by nidas::dynld::raf::SyncServer::openStream(), MergeVerifier::run(), and SensorExtract::run().

void nidas::dynld::SampleInputStream::setMaxSampleTime ( const nidas::util::UTime & val )

inline

See BadSampleFilter.

References _bsf, and nidas::core::BadSampleFilter::setMaxSampleTime().

Referenced by nidas::dynld::raf::SyncServer::openStream(), and MergeVerifier::run().

void nidas::dynld::SampleInputStream::setMinDsmId ( int val )

inline

See BadSampleFilter.

References _bsf, and nidas::core::BadSampleFilter::setMinDsmId().

void nidas::dynld::SampleInputStream::setMinSampleLength ( unsigned int val )

inline

See BadSampleFilter.

References _bsf, and nidas::core::BadSampleFilter::setMinSampleLength().

void nidas::dynld::SampleInputStream::setMinSampleTime ( const nidas::util::UTime & val )

inline

See BadSampleFilter.

References _bsf, and nidas::core::BadSampleFilter::setMinSampleTime().

Referenced by nidas::dynld::raf::SyncServer::openStream(), and MergeVerifier::run().

void SampleInputStream::setNonBlocking	(	bool	val	)
throw	(	nidas::util::IOException
	)

virtual

Implements nidas::core::SampleInput.

xercesc::DOMElement * DOMable::toDOMElement	(	xercesc::DOMElement *	node,
		bool	complete
	)		const
throw	(	xercesc::DOMException
	)

virtualinherited

Add my content into a DOMElement.

Reimplemented in nidas::core::DSMSensor, nidas::core::Variable, nidas::core::ServerSocket, nidas::core::SampleTag, nidas::core::Site, nidas::core::Project, and nidas::core::DSMConfig.

Referenced by nidas::core::ServerSocket::toDOMParent().

xercesc::DOMElement * DOMable::toDOMParent	(	xercesc::DOMElement *	parent,
		bool	complete
	)		const
throw	(	xercesc::DOMException
	)

virtualinherited

Create a DOMElement and append it to the parent.

Reimplemented in nidas::core::DSMSensor, nidas::core::Variable, nidas::core::ServerSocket, nidas::core::SampleTag, nidas::core::Site, nidas::core::Project, and nidas::core::DSMConfig.

Member Data Documentation

bool nidas::dynld::SampleInputStream::_ateof

private

Referenced by read(), and readSamples().

size_t nidas::dynld::SampleInputStream::_badSamples

private

Number of bad samples in the stream so far, which is to say number of bytes checked which did not contain a reasonable sample header.

Referenced by closeBlocks(), getBadSamples(), and sampleFromHeader().

BlockStats nidas::dynld::SampleInputStream::_block

private

Information about the current block of samples, good or bad.

Referenced by closeBlocks(), and sampleFromHeader().

nidas::core::BadSampleFilter nidas::dynld::SampleInputStream::_bsf

private

Referenced by closeBlocks(), parseInputHeader(), sampleFromHeader(), setBadSampleFilter(), setFilterBadSamples(), setMaxDsmId(), setMaxSampleLength(), setMaxSampleTime(), setMinDsmId(), setMinSampleLength(), and setMinSampleTime().

size_t nidas::dynld::SampleInputStream::_dataToRead

private

How many bytes left to read from the stream into the data portion of samp.

Referenced by checkUnexpectedEOF(), and parseInputHeader().

char* nidas::dynld::SampleInputStream::_dptr

private

Pointer into the data portion of samp where we will read next.

const nidas::core::DSMConfig* nidas::dynld::SampleInputStream::_dsm

private

Referenced by getDSMConfig(), and setIOChannel().

nidas::util::EOFException nidas::dynld::SampleInputStream::_eofx

private

Referenced by handleEOF(), and read().

bool nidas::dynld::SampleInputStream::_expectHeader

private

Referenced by getExpectHeader(), parseInputHeader(), and setExpectHeader().

size_t nidas::dynld::SampleInputStream::_goodSamples

private

Number of good samples in the stream so far.

Referenced by closeBlocks(), and sampleFromHeader().

size_t nidas::dynld::SampleInputStream::_headerToRead

private

Referenced by checkUnexpectedEOF(), parseInputHeader(), and sampleFromHeader().

char* nidas::dynld::SampleInputStream::_hptr

private

Referenced by parseInputHeader(), and sampleFromHeader().

nidas::core::SampleInputHeader nidas::dynld::SampleInputStream::_inputHeader

private

Referenced by getInputHeader(), and parseInputHeader().

bool nidas::dynld::SampleInputStream::_inputHeaderParsed

private

Referenced by handleNewInput(), parseInputHeader(), and readInputHeader().

nidas::core::IOChannel* nidas::dynld::SampleInputStream::_iochan

protected

Referenced by close(), getFd(), getName(), isNonBlocking(), SampleInputStream(), setIOChannel(), and ~SampleInputStream().

nidas::core::IOStream* nidas::dynld::SampleInputStream::_iostream

private

Referenced by close(), handleNewInput(), parseInputHeader(), read(), readInputHeader(), readSamples(), sampleFromHeader(), and SampleInputStream().

std::string nidas::dynld::SampleInputStream::_last_name

private

Keep the current name of the input stream so it can be referenced even after the input stream has advanced to a new input name.

Referenced by checkUnexpectedEOF(), closeBlocks(), and readInputHeader().

SampleInputStream* nidas::dynld::SampleInputStream::_original

private

Referenced by getOriginal().

bool nidas::dynld::SampleInputStream::_raw

private

nidas::core::Sample* nidas::dynld::SampleInputStream::_samp

private

Will be non-null if we have previously read part of a sample from the stream.

Referenced by checkUnexpectedEOF(), parseInputHeader(), and ~SampleInputStream().

nidas::core::Sample* nidas::dynld::SampleInputStream::_sampPending

private

The currently pending sample.

When filtering is active, the pending sample is held until the succeeding sample header is confirmed to be good.

Referenced by handleEOF(), and ~SampleInputStream().

nidas::core::DSMService* nidas::dynld::SampleInputStream::_service

private

Service that has requested my input.

nidas::core::SampleHeader nidas::dynld::SampleInputStream::_sheader

private

Referenced by checkUnexpectedEOF(), parseInputHeader(), and sampleFromHeader().

bool nidas::dynld::SampleInputStream::_skipSample

private

This is set if the data for the current sample header should be read but skipped, because the sample is the first after a block of bad samples and thus the time at the front of the sample might still be corrupt.

Referenced by sampleFromHeader().

nidas::core::SampleSourceSupport nidas::dynld::SampleInputStream::_source

protected

Referenced by addSampleClient(), addSampleClientForTag(), addSampleTag(), distribute(), flush(), getClientCount(), getProcessedSampleSource(), getRawSampleSource(), getSampleStats(), getSampleTagIterator(), getSampleTags(), nidas::dynld::isff::PacketInputStream::readSamples(), readSamples(), removeSampleClient(), removeSampleClientForTag(), removeSampleTag(), and setKeepStats().