Creating and Sharing a Synchronized Data Output 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 Output 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 sink of data.

filebuf fbuf; // 1

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

RWByteOutputStream binOutputStream =

RWByteToStreambufOutputStreamImp::make(fbuf); // 2

RWByteOutputStream bufferedBinOutputStream =

RWBufferedByteOutputStreamImp::make(binOutputStream,1024); // 3

RWDataOutputStream dataOutputStream =

RWNativeDataToByteOutputStreamImp::make(bufferedBinOutputStream);// 4

RWDataOutputStream syncDataOutputStream =

RWSynchronizedDataOutputStreamImp::make(dataOutputStream); // 5

try {

// first group of active objects is created

{ synchronizeSingle<double> s1(syncDataOutputStream,64.0); // 6

synchronizeABunch<double> sb1(syncDataOutputStream,3.14159),

sb2(syncDataOutputStream,1.05);

} // 7

// second group of active objects is created

{ synchronizeSingle<int> s1(syncDataOutputStream,67); // 6

synchronizeABunch<int> sb1(syncDataOutputStream,78),

sb2(syncDataOutputStream,99);

} // 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 output 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 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 RWByteToStreambufOutputStreamImp. The class is created by calling its static member function make(), which creates an instance of self and returns it as a binary output stream handle. In this example, the make() function takes a reference to an iostreams streambuf object that is used as the sink of bytes.

//3 Creates an instance of class RWBufferedByteOutputStreamImp. 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 output stream handle. Several static make() functions are available to construct an instance of class RWBufferedByteOutputStreamImp. All take a handle to the next streaming element, which must be of type RWByteOutputStream. 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 RWNativeDataToByteOutputStreamImp. The class is created by calling its static member function make(), which creates an instance of self and returns it as a data output stream handle. The make() function takes a handle to the next streaming element, which must be of type RWByteOutputStream. It is used as the sink for the bytes generated when converting C++ base types to their binary representation.

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

//6 Creates the active objects. They are passed the same synchronized data output stream at construction time. First, an active object writes the same element twenty times. Then, two active objects write the same element ten times in a single synchronized operation. The twenty elements written by the first active object are not necessarily written consecutively into the stream, but each element is written atomically. The elements of the other two active objects are written consecutively into the stream. For example, one element of the first active object might be written into the stream, then the ten elements of the second active object, then the second element of the first active object, then the ten elements of the third active object, and finally the eighteen remaining elements of the first active object.

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

//8 Catches exceptions potentially thrown by the stream. The Streams package defines two exception classes: RWExternalStreamException and RWIncompleteStreamOperation. Class RWExternalStreamException returns an error message and an error code. Class RWIncompleteStreamOperation inherits from class RWExternalStreamException and is thrown when an operation partially succeeds. For more information on exceptions, see Error Handling.