1 /**************************************************************
2 *
3 * Licensed to the Apache Software Foundation (ASF) under one
4 * or more contributor license agreements. See the NOTICE file
5 * distributed with this work for additional information
6 * regarding copyright ownership. The ASF licenses this file
7 * to you under the Apache License, Version 2.0 (the
8 * "License"); you may not use this file except in compliance
9 * with the License. You may obtain a copy of the License at
10 *
11 * http://www.apache.org/licenses/LICENSE-2.0
12 *
13 * Unless required by applicable law or agreed to in writing,
14 * software distributed under the License is distributed on an
15 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16 * KIND, either express or implied. See the License for the
17 * specific language governing permissions and limitations
18 * under the License.
19 *
20 *************************************************************/
21
22
23
24 // MARKER(update_precomp.py): autogen include statement, do not remove
25 #include "precompiled_io.hxx"
26
27 // streams
28 #include <com/sun/star/io/XInputStream.hpp>
29 #include <com/sun/star/io/XOutputStream.hpp>
30 #include <com/sun/star/io/XConnectable.hpp>
31
32 #include <com/sun/star/lang/XServiceInfo.hpp>
33
34 #include <cppuhelper/factory.hxx>
35
36 #include <cppuhelper/implbase4.hxx> // OWeakObject
37
38 #include <osl/conditn.hxx>
39 #include <osl/mutex.hxx>
40
41 #include <limits>
42 #include <string.h>
43
44 using namespace ::rtl;
45 using namespace ::osl;
46 using namespace ::cppu;
47 using namespace ::com::sun::star::uno;
48 using namespace ::com::sun::star::io;
49 using namespace ::com::sun::star::lang;
50
51 #include "factreg.hxx"
52 #include "streamhelper.hxx"
53
54 // Implementation and service names
55 #define IMPLEMENTATION_NAME "com.sun.star.comp.io.stm.Pipe"
56 #define SERVICE_NAME "com.sun.star.io.Pipe"
57
58 namespace io_stm{
59
60 class OPipeImpl :
61 public WeakImplHelper4< XInputStream , XOutputStream , XConnectable , XServiceInfo >
62 {
63 public:
64 OPipeImpl( );
65 ~OPipeImpl();
66
67 public: // XInputStream
68 virtual sal_Int32 SAL_CALL readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead);
69 virtual sal_Int32 SAL_CALL readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead);
70 virtual void SAL_CALL skipBytes(sal_Int32 nBytesToSkip);
71 virtual sal_Int32 SAL_CALL available(void);
72 virtual void SAL_CALL closeInput(void);
73
74 public: // XOutputStream
75
76 virtual void SAL_CALL writeBytes(const Sequence< sal_Int8 >& aData);
77 virtual void SAL_CALL flush(void);
78 virtual void SAL_CALL closeOutput(void);
79
80 public: // XConnectable
81 virtual void SAL_CALL setPredecessor(const Reference< XConnectable >& aPredecessor);
82 virtual Reference< XConnectable > SAL_CALL getPredecessor(void);
83 virtual void SAL_CALL setSuccessor(const Reference < XConnectable > & aSuccessor);
84 virtual Reference < XConnectable > SAL_CALL getSuccessor(void) ;
85
86
87 public: // XServiceInfo
88 OUString SAL_CALL getImplementationName() throw( );
89 Sequence< OUString > SAL_CALL getSupportedServiceNames(void) throw( );
90 sal_Bool SAL_CALL supportsService(const OUString& ServiceName) throw( );
91
92 private:
93
94 // DEBUG
95 inline void checkInvariant();
96
97 Reference < XConnectable > m_succ;
98 Reference < XConnectable > m_pred;
99
100 sal_Int32 m_nBytesToSkip;
101
102 sal_Int8 *m_p;
103
104 sal_Bool m_bOutputStreamClosed;
105 sal_Bool m_bInputStreamClosed;
106
107 oslCondition m_conditionBytesAvail;
108 Mutex m_mutexAccess;
109 IFIFO *m_pFIFO;
110 };
111
112
113
OPipeImpl()114 OPipeImpl::OPipeImpl()
115 {
116 g_moduleCount.modCnt.acquire( &g_moduleCount.modCnt );
117 m_nBytesToSkip = 0;
118
119 m_bOutputStreamClosed = sal_False;
120 m_bInputStreamClosed = sal_False;
121
122 m_pFIFO = new MemFIFO;
123 m_conditionBytesAvail = osl_createCondition();
124 }
125
~OPipeImpl()126 OPipeImpl::~OPipeImpl()
127 {
128 osl_destroyCondition( m_conditionBytesAvail );
129 delete m_pFIFO;
130 g_moduleCount.modCnt.release( &g_moduleCount.modCnt );
131 }
132
133
134 // These invariants must hold when entering a guarded method or leaving a guarded method.
checkInvariant()135 void OPipeImpl::checkInvariant()
136 {
137
138 }
139
readBytes(Sequence<sal_Int8> & aData,sal_Int32 nBytesToRead)140 sal_Int32 OPipeImpl::readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead)
141 {
142 while( sal_True )
143 {
144 { // start guarded section
145 MutexGuard guard( m_mutexAccess );
146 if( m_bInputStreamClosed )
147 {
148 throw NotConnectedException(
149 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::readBytes NotConnectedException" )),
150 *this );
151 }
152 sal_Int32 nOccupiedBufferLen = m_pFIFO->getSize();
153
154 if( m_bOutputStreamClosed && nBytesToRead > nOccupiedBufferLen )
155 {
156 nBytesToRead = nOccupiedBufferLen;
157 }
158
159 if( nOccupiedBufferLen < nBytesToRead )
160 {
161 // wait outside guarded section
162 osl_resetCondition( m_conditionBytesAvail );
163 }
164 else {
165 // necessary bytes are available
166 m_pFIFO->read( aData , nBytesToRead );
167 return nBytesToRead;
168 }
169 } // end guarded section
170
171 // wait for new data outside guarded section!
172 osl_waitCondition( m_conditionBytesAvail , 0 );
173 }
174 }
175
176
readSomeBytes(Sequence<sal_Int8> & aData,sal_Int32 nMaxBytesToRead)177 sal_Int32 OPipeImpl::readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead)
178 {
179 while( sal_True ) {
180 {
181 MutexGuard guard( m_mutexAccess );
182 if( m_bInputStreamClosed )
183 {
184 throw NotConnectedException(
185 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::readSomeBytes NotConnectedException" )),
186 *this );
187 }
188 if( m_pFIFO->getSize() )
189 {
190 sal_Int32 nSize = Min( nMaxBytesToRead , m_pFIFO->getSize() );
191 aData.realloc( nSize );
192 m_pFIFO->read( aData , nSize );
193 return nSize;
194 }
195
196 if( m_bOutputStreamClosed )
197 {
198 // no bytes in buffer anymore
199 return 0;
200 }
201 }
202
203 osl_waitCondition( m_conditionBytesAvail , 0 );
204 }
205 }
206
207
skipBytes(sal_Int32 nBytesToSkip)208 void OPipeImpl::skipBytes(sal_Int32 nBytesToSkip)
209 {
210 MutexGuard guard( m_mutexAccess );
211 if( m_bInputStreamClosed )
212 {
213 throw NotConnectedException(
214 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::skipBytes NotConnectedException" ) ),
215 *this );
216 }
217
218 if( nBytesToSkip < 0
219 || (nBytesToSkip
220 > std::numeric_limits< sal_Int32 >::max() - m_nBytesToSkip) )
221 {
222 throw BufferSizeExceededException(
223 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::skipBytes BufferSizeExceededException" )),
224 *this );
225 }
226 m_nBytesToSkip += nBytesToSkip;
227
228 nBytesToSkip = Min( m_pFIFO->getSize() , m_nBytesToSkip );
229 m_pFIFO->skip( nBytesToSkip );
230 m_nBytesToSkip -= nBytesToSkip;
231 }
232
233
available(void)234 sal_Int32 OPipeImpl::available(void)
235 {
236 MutexGuard guard( m_mutexAccess );
237 if( m_bInputStreamClosed )
238 {
239 throw NotConnectedException(
240 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::available NotConnectedException" ) ),
241 *this );
242 }
243 checkInvariant();
244 return m_pFIFO->getSize();
245 }
246
closeInput(void)247 void OPipeImpl::closeInput(void)
248 {
249 MutexGuard guard( m_mutexAccess );
250
251 m_bInputStreamClosed = sal_True;
252
253 delete m_pFIFO;
254 m_pFIFO = 0;
255
256 // readBytes may throw an exception
257 osl_setCondition( m_conditionBytesAvail );
258
259 setSuccessor( Reference< XConnectable > () );
260 return;
261 }
262
263
writeBytes(const Sequence<sal_Int8> & aData)264 void OPipeImpl::writeBytes(const Sequence< sal_Int8 >& aData)
265 {
266 MutexGuard guard( m_mutexAccess );
267 checkInvariant();
268
269 if( m_bOutputStreamClosed )
270 {
271 throw NotConnectedException(
272 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes NotConnectedException (outputstream)" )),
273 *this );
274 }
275
276 if( m_bInputStreamClosed )
277 {
278 throw NotConnectedException(
279 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes NotConnectedException (inputstream)" )),
280 *this );
281 }
282
283 // check skipping
284 sal_Int32 nLen = aData.getLength();
285 if( m_nBytesToSkip && m_nBytesToSkip >= nLen ) {
286 // all must be skipped - forget whole call
287 m_nBytesToSkip -= nLen;
288 return;
289 }
290
291 // adjust buffersize if necessary
292
293 try
294 {
295 if( m_nBytesToSkip )
296 {
297 Sequence< sal_Int8 > seqCopy( nLen - m_nBytesToSkip );
298 memcpy( seqCopy.getArray() , &( aData.getConstArray()[m_nBytesToSkip] ) , nLen-m_nBytesToSkip );
299 m_pFIFO->write( seqCopy );
300 }
301 else
302 {
303 m_pFIFO->write( aData );
304 }
305 m_nBytesToSkip = 0;
306 }
307 catch ( IFIFO_OutOfBoundsException & )
308 {
309 throw BufferSizeExceededException(
310 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes BufferSizeExceededException" )),
311 *this );
312 }
313 catch ( IFIFO_OutOfMemoryException & )
314 {
315 throw BufferSizeExceededException(
316 OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes BufferSizeExceededException" )),
317 *this );
318 }
319
320 // readBytes may check again if enough bytes are available
321 osl_setCondition( m_conditionBytesAvail );
322
323 checkInvariant();
324 }
325
326
flush(void)327 void OPipeImpl::flush(void)
328 {
329 // nothing to do for a pipe
330 return;
331 }
332
closeOutput(void)333 void OPipeImpl::closeOutput(void)
334 {
335 MutexGuard guard( m_mutexAccess );
336
337 m_bOutputStreamClosed = sal_True;
338 osl_setCondition( m_conditionBytesAvail );
339 setPredecessor( Reference < XConnectable > () );
340 return;
341 }
342
343
setSuccessor(const Reference<XConnectable> & r)344 void OPipeImpl::setSuccessor( const Reference < XConnectable > &r )
345 {
346 /// if the references match, nothing needs to be done
347 if( m_succ != r ) {
348 /// store the reference for later use
349 m_succ = r;
350
351 if( m_succ.is() )
352 {
353 m_succ->setPredecessor(
354 Reference< XConnectable > ( SAL_STATIC_CAST( XConnectable * , this ) ) );
355 }
356 }
357 }
358
getSuccessor()359 Reference < XConnectable > OPipeImpl::getSuccessor()
360 {
361 return m_succ;
362 }
363
364
365 // XDataSource
setPredecessor(const Reference<XConnectable> & r)366 void OPipeImpl::setPredecessor( const Reference < XConnectable > &r )
367 {
368 if( r != m_pred ) {
369 m_pred = r;
370 if( m_pred.is() ) {
371 m_pred->setSuccessor(
372 Reference < XConnectable > ( SAL_STATIC_CAST( XConnectable * , this ) ) );
373 }
374 }
375 }
376
getPredecessor()377 Reference < XConnectable > OPipeImpl::getPredecessor()
378 {
379 return m_pred;
380 }
381
382
383
384
385 // XServiceInfo
getImplementationName()386 OUString OPipeImpl::getImplementationName() throw( )
387 {
388 return OPipeImpl_getImplementationName();
389 }
390
391 // XServiceInfo
supportsService(const OUString & ServiceName)392 sal_Bool OPipeImpl::supportsService(const OUString& ServiceName) throw( )
393 {
394 Sequence< OUString > aSNL = getSupportedServiceNames();
395 const OUString * pArray = aSNL.getConstArray();
396
397 for( sal_Int32 i = 0; i < aSNL.getLength(); i++ )
398 if( pArray[i] == ServiceName )
399 return sal_True;
400
401 return sal_False;
402 }
403
404 // XServiceInfo
getSupportedServiceNames(void)405 Sequence< OUString > OPipeImpl::getSupportedServiceNames(void) throw( )
406 {
407 return OPipeImpl_getSupportedServiceNames();
408 }
409
410
411
412
413
414 /* implementation functions
415 *
416 *
417 */
418
419
OPipeImpl_CreateInstance(const Reference<XComponentContext> &)420 Reference < XInterface > SAL_CALL OPipeImpl_CreateInstance(
421 const Reference < XComponentContext > & )
422 {
423 OPipeImpl *p = new OPipeImpl;
424
425 return Reference < XInterface > ( SAL_STATIC_CAST( OWeakObject * , p ) );
426 }
427
428
OPipeImpl_getImplementationName()429 OUString OPipeImpl_getImplementationName()
430 {
431 return OUString( RTL_CONSTASCII_USTRINGPARAM ( IMPLEMENTATION_NAME ) );
432 }
433
OPipeImpl_getSupportedServiceNames(void)434 Sequence<OUString> OPipeImpl_getSupportedServiceNames(void)
435 {
436 Sequence<OUString> aRet(1);
437 aRet.getArray()[0] = OUString( RTL_CONSTASCII_USTRINGPARAM( SERVICE_NAME ));
438 return aRet;
439 }
440 }
441