Creating and Sharing a Synchronized Data Input Stream Among Several Active Objects

SourcePro Core : Advanced Tools Module User’s Guide : PART II Advanced Tools Module Packages : Using Streams : Examples : Creating and Using Thread-safe Chains of Streaming Elements : Creating and Sharing a Synchronized Data Input Stream Among Several Active Objects

The following code constructs the thread-safe chain of streaming elements, and then constructs several active objects that use the same chain of streaming elements as the source of data.

filebuf fbuf; // 1

fbuf.open("dataFilteredWrite.dat", ios::in | ios::binary);

RWByteInputStream binInputStream =

RWByteFromStreambufInputStreamImp::make(fbuf); // 2

RWByteInputStream bufferedBinInputStream =

RWBufferedByteInputStreamImp::make(binInputStream,1024); // 3

RWDataInputStream dataInputStream =

RWNativeDataFromByteInputStreamImp::make(bufferedBinInputStream);// 4

RWDataInputStream syncDataInputStream =

RWSynchronizedDataInputStreamImp::make(dataInputStream); // 5

try {

{

readABunch<double> sb1(syncDataInputStream),

sb2(syncDataInputStream), // 6

sb3(syncDataInputStream),

sb4(syncDataInputStream);

} // 7

{

readABunch<int> sb5(syncDataInputStream),

sb6(syncDataInputStream), // 6

sb7(syncDataInputStream),

sb8(syncDataInputStream);

} // 7

}

catch(const RWIncompleteStreamOperation& e) { // 8

cout << e.why() << endl;

cout << e.elementsProcessed() << endl;

}

catch(const RWExternalStreamException& e) {

cout << e.why() << endl;

}

//1 Creates an iostreams file buffer, and opens it in input mode. If the code is compiled on a PC platform, the file buffer needs to be opened in binary mode by specifying the flag ios::binary. If you built the Advanced Tools Module with the Standard iostreams library, you need to qualify filebuf and other iostreams elements with std::, or you need to include using declarations. The complete example code makes use of macros defined by the Essential Tools Module in order to support both the classic and Standard iostreams.

//2 Creates an instance of class RWByteFromStreambufInputStreamImp. The class is created by calling its static member function make(), which creates an instance of self and returns it as a binary input stream handle. In this example, the make() function takes a reference to an iostreams streambuf object that is used as the source of bytes.

//3 Creates an instance of class RWBufferedByteInputStreamImp. The class is created by calling one of its static member functions make(), which creates an instance of self and returns it as a binary input stream handle. Several static make() functions are available to construct an instance of class RWBufferedByteInputStreamImp. All of them take a handle to the next streaming element, which must be of type RWByteInputStream. The static make() function used in this example takes a second parameter that specifies the size of the buffer to be allocated.

//4 Creates an instance of class RWNativeDataFromByteInputStreamImp. The class is created by calling its static member function make(), which creates an instance of self and returns it as a data input stream handle. The make() function takes a handle to the next streaming element, which must be of type RWByteInputStream. It is used as the source of bytes.

//5 Creates an instance of class RWSynchronizedDataInputStreamImp. The class is created by calling its static member function make(), which creates an instance of self and returns it as a data input stream handle. The make() function takes a handle to the next streaming element, which must be of type RWDataInputStream. Each operation executed on this stream is guaranteed to be synchronized in a multithreaded environment.

//6 Creates four active objects. They are passed the same synchronized data input stream at construction time. Each active object reads ten consecutive elements from the stream, and sends them to the standard output.

//7 Each active object waits for its thread to finish execution before being destroyed.

//8 Catches exceptions potentially thrown by the stream. Class RWExternalStreamException is the base class for all the exceptions thrown by the Streams package. It returns an error message and an error code. For more information on exceptions, see Error Handling.