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