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 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 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. 135 void OPipeImpl::checkInvariant() 136 { 137 138 } 139 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 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 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 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 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 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 327 void OPipeImpl::flush(void) 328 { 329 // nothing to do for a pipe 330 return; 331 } 332 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 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 359 Reference < XConnectable > OPipeImpl::getSuccessor() 360 { 361 return m_succ; 362 } 363 364 365 // XDataSource 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 377 Reference < XConnectable > OPipeImpl::getPredecessor() 378 { 379 return m_pred; 380 } 381 382 383 384 385 // XServiceInfo 386 OUString OPipeImpl::getImplementationName() throw( ) 387 { 388 return OPipeImpl_getImplementationName(); 389 } 390 391 // XServiceInfo 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 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 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 429 OUString OPipeImpl_getImplementationName() 430 { 431 return OUString( RTL_CONSTASCII_USTRINGPARAM ( IMPLEMENTATION_NAME ) ); 432 } 433 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