io/source/stm/omark.cxx

   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 #include <map>
  22 #include <memory>
  23 #include <optional>
  24
  25 #include <com/sun/star/io/BufferSizeExceededException.hpp>
  26 #include <com/sun/star/io/NotConnectedException.hpp>
  27 #include <com/sun/star/io/XMarkableStream.hpp>
  28 #include <com/sun/star/io/XOutputStream.hpp>
  29 #include <com/sun/star/io/XInputStream.hpp>
  30 #include <com/sun/star/io/XActiveDataSource.hpp>
  31 #include <com/sun/star/io/XActiveDataSink.hpp>
  32 #include <com/sun/star/io/XConnectable.hpp>
  33 #include <com/sun/star/lang/IllegalArgumentException.hpp>
  34 #include <com/sun/star/lang/XServiceInfo.hpp>
  35 #include <com/sun/star/uno/XComponentContext.hpp>
  36
  37 #include <cppuhelper/weak.hxx>
  38 #include <cppuhelper/implbase.hxx>
  39 #include <cppuhelper/supportsservice.hxx>
  40
  41 #include <osl/diagnose.h>
  42 #include <mutex>
  43
  44 using namespace ::cppu;
  45 using namespace ::com::sun::star::io;
  46 using namespace ::com::sun::star::uno;
  47 using namespace ::com::sun::star::lang;
  48
  49 #include "streamhelper.hxx"
  50
  51 namespace io_stm {
  52
  53 namespace {
  54
  55 /***********************
  56 *
  57 * OMarkableOutputStream.
  58 *
  59 * This object allows to set marks in an outputstream. It is allowed to jump back to the marks and
  60 * rewrite the same bytes.
  61 *
  62 *         The object must buffer the data since the last mark set. Flush will not
  63 *         have any effect. As soon as the last mark has been removed, the object may write the data
  64 *         through to the chained object.
  65 *
  66 **********************/
  67 class OMarkableOutputStream :
  68     public WeakImplHelper< XOutputStream ,
  69                             XActiveDataSource ,
  70                             XMarkableStream ,
  71                             XConnectable,
  72                             XServiceInfo
  73                           >
  74 {
  75 public:
  76     OMarkableOutputStream(  );
  77
  78 public: // XOutputStream
  79     virtual void SAL_CALL writeBytes(const Sequence< sal_Int8 >& aData) override;
  80     virtual void SAL_CALL flush() override;
  81     virtual void SAL_CALL closeOutput() override;
  82
  83 public: // XMarkable
  84     virtual sal_Int32 SAL_CALL createMark() override;
  85     virtual void SAL_CALL deleteMark(sal_Int32 Mark) override;
  86     virtual void SAL_CALL jumpToMark(sal_Int32 nMark) override;
  87     virtual void SAL_CALL jumpToFurthest() override;
  88     virtual sal_Int32 SAL_CALL offsetToMark(sal_Int32 nMark) override;
  89
  90 public: // XActiveDataSource
  91     virtual void SAL_CALL setOutputStream(const Reference < XOutputStream > & aStream) override;
  92     virtual Reference < XOutputStream > SAL_CALL getOutputStream() override;
  93
  94 public: // XConnectable
  95     virtual void SAL_CALL setPredecessor(const Reference < XConnectable > & aPredecessor) override;
  96     virtual Reference < XConnectable > SAL_CALL getPredecessor() override;
  97     virtual void SAL_CALL setSuccessor(const Reference < XConnectable >& aSuccessor) override;
  98     virtual Reference<  XConnectable >  SAL_CALL getSuccessor() override;
  99
 100 public: // XServiceInfo
 101     OUString                     SAL_CALL getImplementationName() override;
 102     Sequence< OUString >         SAL_CALL getSupportedServiceNames() override;
 103     sal_Bool                        SAL_CALL supportsService(const OUString& ServiceName) override;
 104
 105 private:
 106     // helper methods
 107     /// @throws NotConnectedException
 108     /// @throws BufferSizeExceededException
 109     void checkMarksAndFlush();
 110
 111     Reference< XConnectable > m_succ;
 112     Reference< XConnectable > m_pred;
 113
 114     Reference< XOutputStream >  m_output;
 115     bool m_bValidStream;
 116
 117     MemRingBuffer m_aRingBuffer;
 118     std::map<sal_Int32,sal_Int32,std::less< sal_Int32 > > m_mapMarks;
 119     sal_Int32 m_nCurrentPos;
 120     sal_Int32 m_nCurrentMark;
 121
 122     std::mutex m_mutex;
 123 };
 124
 125 }
 126
 127 OMarkableOutputStream::OMarkableOutputStream( )
 128     : m_bValidStream(false)
 129     , m_nCurrentPos(0)
 130     , m_nCurrentMark(0)
 131 {
 132 }
 133
 134 // XOutputStream
 135 void OMarkableOutputStream::writeBytes(const Sequence< sal_Int8 >& aData)
 136 {
 137     std::unique_lock guard( m_mutex );
 138
 139     if( !m_bValidStream ) {
 140         throw NotConnectedException();
 141     }
 142     if( m_mapMarks.empty() && ( m_aRingBuffer.getSize() == 0 ) ) {
 143         // no mark and  buffer active, simple write through
 144         m_output->writeBytes( aData );
 145     }
 146     else {
 147         // new data must be buffered
 148         m_aRingBuffer.writeAt( m_nCurrentPos , aData );
 149         m_nCurrentPos += aData.getLength();
 150         checkMarksAndFlush();
 151     }
 152
 153 }
 154
 155 void OMarkableOutputStream::flush()
 156 {
 157     Reference< XOutputStream > output;
 158     {
 159         std::unique_lock guard( m_mutex );
 160         output = m_output;
 161     }
 162
 163     // Markable cannot flush buffered data, because the data may get rewritten,
 164     // however one can forward the flush to the chained stream to give it
 165     // a chance to write data buffered in the chained stream.
 166     if( output.is() )
 167     {
 168         output->flush();
 169     }
 170 }
 171
 172 void OMarkableOutputStream::closeOutput()
 173 {
 174     if( !m_bValidStream ) {
 175         throw NotConnectedException();
 176     }
 177     std::unique_lock guard( m_mutex );
 178     // all marks must be cleared and all
 179
 180     m_mapMarks.clear();
 181     m_nCurrentPos = m_aRingBuffer.getSize();
 182     checkMarksAndFlush();
 183
 184     m_output->closeOutput();
 185
 186     setOutputStream( Reference< XOutputStream > () );
 187     setPredecessor( Reference < XConnectable >() );
 188     setSuccessor( Reference< XConnectable > () );
 189
 190 }
 191
 192
 193 sal_Int32 OMarkableOutputStream::createMark()
 194 {
 195     std::unique_lock guard( m_mutex );
 196     sal_Int32 nMark = m_nCurrentMark;
 197
 198     m_mapMarks[nMark] = m_nCurrentPos;
 199
 200     m_nCurrentMark ++;
 201     return nMark;
 202 }
 203
 204 void OMarkableOutputStream::deleteMark(sal_Int32 Mark)
 205 {
 206     std::unique_lock guard( m_mutex );
 207     std::map<sal_Int32,sal_Int32,std::less<sal_Int32> >::iterator ii = m_mapMarks.find( Mark );
 208
 209     if( ii == m_mapMarks.end() ) {
 210         throw IllegalArgumentException(
 211             "MarkableOutputStream::deleteMark unknown mark (" + OUString::number(Mark) + ")",
 212             *this, 0);
 213     }
 214     m_mapMarks.erase( ii );
 215     checkMarksAndFlush();
 216 }
 217
 218 void OMarkableOutputStream::jumpToMark(sal_Int32 nMark)
 219 {
 220     std::unique_lock guard( m_mutex );
 221     std::map<sal_Int32,sal_Int32,std::less<sal_Int32> >::iterator ii = m_mapMarks.find( nMark );
 222
 223     if( ii == m_mapMarks.end() ) {
 224         throw IllegalArgumentException(
 225             "MarkableOutputStream::jumpToMark unknown mark (" + OUString::number(nMark) + ")",
 226             *this, 0);
 227     }
 228     m_nCurrentPos = (*ii).second;
 229 }
 230
 231 void OMarkableOutputStream::jumpToFurthest()
 232 {
 233     std::unique_lock guard( m_mutex );
 234     m_nCurrentPos = m_aRingBuffer.getSize();
 235     checkMarksAndFlush();
 236 }
 237
 238 sal_Int32 OMarkableOutputStream::offsetToMark(sal_Int32 nMark)
 239 {
 240
 241     std::unique_lock guard( m_mutex );
 242     std::map<sal_Int32,sal_Int32,std::less<sal_Int32> >::const_iterator ii = m_mapMarks.find( nMark );
 243
 244     if( ii == m_mapMarks.end() )
 245     {
 246         throw IllegalArgumentException(
 247             "MarkableOutputStream::offsetToMark unknown mark (" + OUString::number(nMark) + ")",
 248             *this, 0);
 249     }
 250     return m_nCurrentPos - (*ii).second;
 251 }
 252
 253
 254 // XActiveDataSource2
 255 void OMarkableOutputStream::setOutputStream(const Reference < XOutputStream >& aStream)
 256 {
 257     if( m_output != aStream ) {
 258         m_output = aStream;
 259
 260         Reference < XConnectable > succ( m_output , UNO_QUERY );
 261         setSuccessor( succ );
 262     }
 263     m_bValidStream = m_output.is();
 264 }
 265
 266 Reference< XOutputStream > OMarkableOutputStream::getOutputStream()
 267 {
 268     return m_output;
 269 }
 270
 271
 272 void OMarkableOutputStream::setSuccessor( const Reference< XConnectable > &r )
 273 {
 274      /// if the references match, nothing needs to be done
 275      if( m_succ != r ) {
 276          /// store the reference for later use
 277          m_succ = r;
 278
 279          if( m_succ.is() ) {
 280               m_succ->setPredecessor( Reference < XConnectable > (
 281                   static_cast< XConnectable *  >(this) ) );
 282          }
 283      }
 284 }
 285 Reference <XConnectable > OMarkableOutputStream::getSuccessor()
 286 {
 287     return m_succ;
 288 }
 289
 290
 291 // XDataSource
 292 void OMarkableOutputStream::setPredecessor( const Reference< XConnectable > &r )
 293 {
 294     if( r != m_pred ) {
 295         m_pred = r;
 296         if( m_pred.is() ) {
 297             m_pred->setSuccessor( Reference < XConnectable > (
 298                 static_cast< XConnectable *  >(this ) ) );
 299         }
 300     }
 301 }
 302 Reference < XConnectable > OMarkableOutputStream::getPredecessor()
 303 {
 304     return m_pred;
 305 }
 306
 307
 308 // private methods
 309
 310 void OMarkableOutputStream::checkMarksAndFlush()
 311 {
 312     // find the smallest mark
 313     sal_Int32 nNextFound = m_nCurrentPos;
 314     for (auto const& mark : m_mapMarks)
 315     {
 316         if( mark.second <= nNextFound )  {
 317             nNextFound = mark.second;
 318         }
 319     }
 320
 321     if( nNextFound ) {
 322         // some data must be released !
 323         m_nCurrentPos -= nNextFound;
 324         for (auto & mark : m_mapMarks)
 325         {
 326             mark.second -= nNextFound;
 327         }
 328
 329         Sequence<sal_Int8> seq(nNextFound);
 330         m_aRingBuffer.readAt( 0 , seq , nNextFound );
 331         m_aRingBuffer.forgetFromStart( nNextFound );
 332
 333         // now write data through to streams
 334         m_output->writeBytes( seq );
 335     }
 336     else {
 337         // nothing to do. There is a mark or the current cursor position, that prevents
 338         // releasing data !
 339     }
 340 }
 341
 342
 343 // XServiceInfo
 344 OUString OMarkableOutputStream::getImplementationName()
 345 {
 346     return u"com.sun.star.comp.io.stm.MarkableOutputStream"_ustr;
 347 }
 348
 349 // XServiceInfo
 350 sal_Bool OMarkableOutputStream::supportsService(const OUString& ServiceName)
 351 {
 352     return cppu::supportsService(this, ServiceName);
 353 }
 354
 355 // XServiceInfo
 356 Sequence< OUString > OMarkableOutputStream::getSupportedServiceNames()
 357 {
 358     return { u"com.sun.star.io.MarkableOutputStream"_ustr };
 359 }
 360
 361 extern "C" SAL_DLLPUBLIC_EXPORT css::uno::XInterface*
 362 io_OMarkableOutputStream_get_implementation(
 363     css::uno::XComponentContext* , css::uno::Sequence<css::uno::Any> const&)
 364 {
 365     return cppu::acquire(new OMarkableOutputStream());
 366 }
 367
 368
 369 // XMarkableInputStream
 370
 371 namespace {
 372
 373 class OMarkableInputStream :
 374     public WeakImplHelper
 375     <
 376              XInputStream,
 377              XActiveDataSink,
 378              XMarkableStream,
 379              XConnectable,
 380              XServiceInfo
 381     >
 382 {
 383 public:
 384     OMarkableInputStream(  );
 385
 386
 387 public: // XInputStream
 388     virtual sal_Int32 SAL_CALL readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead) override ;
 389     virtual sal_Int32 SAL_CALL readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead) override;
 390     virtual void SAL_CALL skipBytes(sal_Int32 nBytesToSkip) override;
 391
 392     virtual sal_Int32 SAL_CALL available() override;
 393     virtual void SAL_CALL closeInput() override;
 394
 395 public: // XMarkable
 396     virtual sal_Int32 SAL_CALL createMark() override;
 397     virtual void SAL_CALL deleteMark(sal_Int32 Mark) override;
 398     virtual void SAL_CALL jumpToMark(sal_Int32 nMark) override;
 399     virtual void SAL_CALL jumpToFurthest() override;
 400     virtual sal_Int32 SAL_CALL offsetToMark(sal_Int32 nMark) override;
 401
 402 public: // XActiveDataSink
 403     virtual void SAL_CALL setInputStream(const Reference < XInputStream > & aStream) override;
 404     virtual Reference < XInputStream > SAL_CALL getInputStream() override;
 405
 406 public: // XConnectable
 407     virtual void SAL_CALL setPredecessor(const Reference < XConnectable > & aPredecessor) override;
 408     virtual Reference < XConnectable > SAL_CALL getPredecessor() override;
 409     virtual void SAL_CALL setSuccessor(const Reference < XConnectable > & aSuccessor) override;
 410     virtual Reference < XConnectable > SAL_CALL getSuccessor() override;
 411
 412 public: // XServiceInfo
 413     OUString                     SAL_CALL getImplementationName() override;
 414     Sequence< OUString >         SAL_CALL getSupportedServiceNames() override;
 415     sal_Bool                         SAL_CALL  supportsService(const OUString& ServiceName) override;
 416
 417 private:
 418     void checkMarksAndFlush();
 419
 420     Reference < XConnectable >  m_succ;
 421     Reference < XConnectable >  m_pred;
 422
 423     Reference< XInputStream > m_input;
 424     bool m_bValidStream;
 425
 426     std::optional<MemRingBuffer> m_oBuffer;
 427     std::map<sal_Int32,sal_Int32,std::less< sal_Int32 > > m_mapMarks;
 428     sal_Int32 m_nCurrentPos;
 429     sal_Int32 m_nCurrentMark;
 430
 431     std::mutex m_mutex;
 432 };
 433
 434 }
 435
 436 OMarkableInputStream::OMarkableInputStream()
 437     : m_bValidStream(false)
 438     , m_nCurrentPos(0)
 439     , m_nCurrentMark(0)
 440 {
 441     m_oBuffer.emplace();
 442 }
 443
 444
 445 // XInputStream
 446
 447 sal_Int32 OMarkableInputStream::readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead)
 448 {
 449     std::unique_lock guard( m_mutex );
 450
 451     if( !m_bValidStream ) {
 452         throw NotConnectedException(
 453             u"MarkableInputStream::readBytes NotConnectedException"_ustr,
 454             *this );
 455     }
 456
 457     sal_Int32 nBytesRead;
 458     if( m_mapMarks.empty() && ! m_oBuffer->getSize() ) {
 459         // normal read !
 460         nBytesRead = m_input->readBytes( aData, nBytesToRead );
 461     }
 462     else {
 463         // read from buffer
 464         sal_Int32 nRead;
 465
 466         // read enough bytes into buffer
 467         if( m_oBuffer->getSize() - m_nCurrentPos < nBytesToRead  ) {
 468             sal_Int32 nToRead = nBytesToRead - ( m_oBuffer->getSize() - m_nCurrentPos );
 469             nRead = m_input->readBytes( aData , nToRead );
 470
 471             OSL_ASSERT( aData.getLength() == nRead );
 472
 473             m_oBuffer->writeAt( m_oBuffer->getSize() , aData );
 474
 475             if( nRead < nToRead ) {
 476                 nBytesToRead = nBytesToRead - (nToRead-nRead);
 477             }
 478         }
 479
 480         OSL_ASSERT( m_oBuffer->getSize() - m_nCurrentPos >= nBytesToRead  );
 481
 482         m_oBuffer->readAt( m_nCurrentPos , aData , nBytesToRead );
 483
 484         m_nCurrentPos += nBytesToRead;
 485         nBytesRead = nBytesToRead;
 486     }
 487
 488     return nBytesRead;
 489 }
 490
 491
 492 sal_Int32 OMarkableInputStream::readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead)
 493 {
 494     std::unique_lock guard( m_mutex );
 495
 496     if( !m_bValidStream )    {
 497         throw NotConnectedException(
 498             u"MarkableInputStream::readSomeBytes NotConnectedException"_ustr,
 499             *this );
 500     }
 501
 502     sal_Int32 nBytesRead;
 503     if( m_mapMarks.empty() && ! m_oBuffer->getSize() ) {
 504         // normal read !
 505         nBytesRead = m_input->readSomeBytes( aData, nMaxBytesToRead );
 506     }
 507     else {
 508         // read from buffer
 509         sal_Int32 nRead = 0;
 510         sal_Int32 nInBuffer = m_oBuffer->getSize() - m_nCurrentPos;
 511         sal_Int32 nAdditionalBytesToRead = std::min<sal_Int32>(nMaxBytesToRead-nInBuffer,m_input->available());
 512         nAdditionalBytesToRead = std::max<sal_Int32>(0 , nAdditionalBytesToRead );
 513
 514         // read enough bytes into buffer
 515         if( 0 == nInBuffer ) {
 516             nRead = m_input->readSomeBytes( aData , nMaxBytesToRead );
 517         }
 518         else if( nAdditionalBytesToRead ) {
 519             nRead = m_input->readBytes( aData , nAdditionalBytesToRead );
 520         }
 521
 522         if( nRead ) {
 523             aData.realloc( nRead );
 524             m_oBuffer->writeAt( m_oBuffer->getSize() , aData );
 525         }
 526
 527         nBytesRead = std::min( nMaxBytesToRead , nInBuffer + nRead );
 528
 529         // now take everything from buffer !
 530         m_oBuffer->readAt( m_nCurrentPos , aData , nBytesRead );
 531
 532         m_nCurrentPos += nBytesRead;
 533     }
 534
 535     return nBytesRead;
 536
 537
 538 }
 539
 540
 541 void OMarkableInputStream::skipBytes(sal_Int32 nBytesToSkip)
 542 {
 543     if ( nBytesToSkip < 0 )
 544         throw BufferSizeExceededException(
 545             u"precondition not met: XInputStream::skipBytes: non-negative integer required!"_ustr,
 546             *this
 547         );
 548
 549     // this method is blocking
 550     Sequence<sal_Int8> seqDummy( nBytesToSkip );
 551     readBytes( seqDummy , nBytesToSkip );
 552 }
 553
 554 sal_Int32 OMarkableInputStream::available()
 555 {
 556     std::unique_lock guard( m_mutex );
 557
 558     if( !m_bValidStream )    {
 559         throw NotConnectedException(
 560             u"MarkableInputStream::available NotConnectedException"_ustr,
 561             *this );
 562     }
 563
 564     sal_Int32 nAvail = m_input->available() + ( m_oBuffer->getSize() - m_nCurrentPos );
 565     return nAvail;
 566 }
 567
 568
 569 void OMarkableInputStream::closeInput()
 570 {
 571     std::unique_lock guard( m_mutex );
 572
 573     if( !m_bValidStream ) {
 574         throw NotConnectedException(
 575             u"MarkableInputStream::closeInput NotConnectedException"_ustr,
 576             *this );
 577     }
 578
 579     m_input->closeInput();
 580
 581     m_input.clear();
 582     if( m_pred )
 583         m_pred.clear();
 584     if( m_succ )
 585         m_succ.clear();
 586     m_bValidStream = false;
 587     m_oBuffer.reset();
 588     m_nCurrentPos = 0;
 589     m_nCurrentMark = 0;
 590 }
 591
 592 // XMarkable
 593
 594 sal_Int32 OMarkableInputStream::createMark()
 595 {
 596     std::unique_lock guard( m_mutex );
 597     sal_Int32 nMark = m_nCurrentMark;
 598
 599     m_mapMarks[nMark] = m_nCurrentPos;
 600
 601     m_nCurrentMark ++;
 602     return nMark;
 603 }
 604
 605 void OMarkableInputStream::deleteMark(sal_Int32 Mark)
 606 {
 607     std::unique_lock guard( m_mutex );
 608     std::map<sal_Int32,sal_Int32,std::less<sal_Int32> >::iterator ii = m_mapMarks.find( Mark );
 609
 610     if( ii == m_mapMarks.end() ) {
 611         throw IllegalArgumentException(
 612             "MarkableInputStream::deleteMark unknown mark (" + OUString::number(Mark) + ")",
 613             *this , 0 );
 614     }
 615     m_mapMarks.erase( ii );
 616     checkMarksAndFlush();
 617 }
 618
 619 void OMarkableInputStream::jumpToMark(sal_Int32 nMark)
 620 {
 621     std::unique_lock guard( m_mutex );
 622     std::map<sal_Int32,sal_Int32,std::less<sal_Int32> >::iterator ii = m_mapMarks.find( nMark );
 623
 624     if( ii == m_mapMarks.end() )
 625     {
 626         throw IllegalArgumentException(
 627             "MarkableInputStream::jumpToMark unknown mark (" + OUString::number(nMark) + ")",
 628             *this , 0 );
 629     }
 630     m_nCurrentPos = (*ii).second;
 631 }
 632
 633 void OMarkableInputStream::jumpToFurthest()
 634 {
 635     std::unique_lock guard( m_mutex );
 636     m_nCurrentPos = m_oBuffer->getSize();
 637     checkMarksAndFlush();
 638 }
 639
 640 sal_Int32 OMarkableInputStream::offsetToMark(sal_Int32 nMark)
 641 {
 642     std::unique_lock guard( m_mutex );
 643     std::map<sal_Int32,sal_Int32,std::less<sal_Int32> >::const_iterator ii = m_mapMarks.find( nMark );
 644
 645     if( ii == m_mapMarks.end() )
 646     {
 647         throw IllegalArgumentException(
 648             "MarkableInputStream::offsetToMark unknown mark (" + OUString::number(nMark) + ")",
 649             *this, 0 );
 650     }
 651     return m_nCurrentPos - (*ii).second;
 652 }
 653
 654
 655 // XActiveDataSource
 656 void OMarkableInputStream::setInputStream(const Reference< XInputStream > & aStream)
 657 {
 658     Reference < XConnectable > pred;
 659     {
 660         std::unique_lock guard( m_mutex );
 661         if( m_input == aStream )
 662             return;
 663
 664         m_input = aStream;
 665         m_bValidStream = m_input.is();
 666         pred.set( m_input , UNO_QUERY );
 667     }
 668     setPredecessor( pred );
 669 }
 670
 671 Reference< XInputStream > OMarkableInputStream::getInputStream()
 672 {
 673     std::unique_lock guard( m_mutex );
 674     return m_input;
 675 }
 676
 677
 678 // XDataSink
 679 void OMarkableInputStream::setSuccessor( const Reference< XConnectable > &r )
 680 {
 681     {
 682         std::unique_lock guard( m_mutex );
 683         /// if the references match, nothing needs to be done
 684         if( m_succ == r )
 685             return;
 686
 687         /// store the reference for later use
 688         m_succ = r;
 689     }
 690     if( r ) {
 691         /// set this instance as the sink !
 692         r->setPredecessor( Reference< XConnectable > ( static_cast< XConnectable * >(this) ) );
 693     }
 694 }
 695
 696 Reference < XConnectable >  OMarkableInputStream::getSuccessor()
 697 {
 698     std::unique_lock guard( m_mutex );
 699     return m_succ;
 700 }
 701
 702
 703 // XDataSource
 704 void OMarkableInputStream::setPredecessor( const Reference < XConnectable >  &r )
 705 {
 706     {
 707         std::unique_lock guard( m_mutex );
 708         if( r == m_pred )
 709             return;
 710         m_pred = r;
 711     }
 712     if( r ) {
 713         r->setSuccessor( Reference< XConnectable > (
 714             static_cast< XConnectable * >(this) ) );
 715     }
 716 }
 717
 718 Reference< XConnectable >  OMarkableInputStream::getPredecessor()
 719 {
 720     std::unique_lock guard( m_mutex );
 721     return m_pred;
 722 }
 723
 724
 725 void OMarkableInputStream::checkMarksAndFlush()
 726 {
 727     // find the smallest mark
 728     sal_Int32 nNextFound = m_nCurrentPos;
 729     for (auto const& mark : m_mapMarks)
 730     {
 731         if( mark.second <= nNextFound )  {
 732             nNextFound = mark.second;
 733         }
 734     }
 735
 736     if( nNextFound ) {
 737         // some data must be released !
 738         m_nCurrentPos -= nNextFound;
 739         for (auto & mark : m_mapMarks)
 740         {
 741             mark.second -= nNextFound;
 742         }
 743
 744         m_oBuffer->forgetFromStart( nNextFound );
 745
 746     }
 747     else {
 748         // nothing to do. There is a mark or the current cursor position, that prevents
 749         // releasing data !
 750     }
 751 }
 752
 753 // XServiceInfo
 754 OUString OMarkableInputStream::getImplementationName()
 755 {
 756     return u"com.sun.star.comp.io.stm.MarkableInputStream"_ustr;
 757 }
 758
 759 // XServiceInfo
 760 sal_Bool OMarkableInputStream::supportsService(const OUString& ServiceName)
 761 {
 762     return cppu::supportsService(this, ServiceName);
 763 }
 764
 765 // XServiceInfo
 766 Sequence< OUString > OMarkableInputStream::getSupportedServiceNames()
 767 {
 768     return { u"com.sun.star.io.MarkableInputStream"_ustr };
 769 }
 770
 771 extern "C" SAL_DLLPUBLIC_EXPORT css::uno::XInterface*
 772 io_OMarkableInputStream_get_implementation(
 773     css::uno::XComponentContext* , css::uno::Sequence<css::uno::Any> const&)
 774 {
 775     return cppu::acquire(new OMarkableInputStream());
 776 }
 777
 778 }
 779
 780 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */