sc/source/ui/docshell/datastream.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
  10 #include <datastream.hxx>
  11 #include <datastreamgettime.hxx>
  12
  13 #include <com/sun/star/frame/XLayoutManager.hpp>
  14 #include <osl/conditn.hxx>
  15 #include <osl/time.h>
  16 #include <salhelper/thread.hxx>
  17 #include <sfx2/viewfrm.hxx>
  18 #include <tools/stream.hxx>
  19 #include <vcl/svapp.hxx>
  20 #include <docsh.hxx>
  21 #include <tabvwsh.hxx>
  22 #include <viewdata.hxx>
  23 #include <stringutil.hxx>
  24 #include <documentlinkmgr.hxx>
  25 #include <o3tl/enumarray.hxx>
  26
  27 #include <officecfg/Office/Calc.hxx>
  28
  29 #include <orcus/csv_parser.hpp>
  30
  31 #include <atomic>
  32 #include <queue>
  33
  34 namespace com::sun::star::ui { class XUIElement; }
  35
  36 namespace sc {
  37
  38 static o3tl::enumarray<DebugTime, double> fTimes { 0.0, 0.0, 0.0 };
  39
  40 double datastream_get_time(DebugTime nIdx)
  41 {
  42     return fTimes[ nIdx ];
  43 }
  44
  45 namespace {
  46
  47 double getNow()
  48 {
  49     TimeValue now;
  50     osl_getSystemTime(&now);
  51     return static_cast<double>(now.Seconds) + static_cast<double>(now.Nanosec) / 1000000000.0;
  52 }
  53
  54 class CSVHandler
  55 {
  56     DataStream::Line& mrLine;
  57     size_t mnColCount;
  58     size_t mnCols;
  59     const char* mpLineHead;
  60
  61 public:
  62     CSVHandler( DataStream::Line& rLine, size_t nColCount ) :
  63         mrLine(rLine), mnColCount(nColCount), mnCols(0), mpLineHead(rLine.maLine.getStr()) {}
  64
  65     static void begin_parse() {}
  66     static void end_parse() {}
  67     static void begin_row() {}
  68     static void end_row() {}
  69
  70     void cell(std::string_view s, bool /*transient*/)
  71     {
  72         if (mnCols >= mnColCount)
  73             return;
  74
  75         DataStream::Cell aCell;
  76         if (ScStringUtil::parseSimpleNumber(s.data(), s.size(), '.', ',', aCell.mfValue))
  77         {
  78             aCell.mbValue = true;
  79         }
  80         else
  81         {
  82             aCell.mbValue = false;
  83             aCell.maStr.Pos = std::distance(mpLineHead, s.data());
  84             aCell.maStr.Size = s.size();
  85         }
  86         mrLine.maCells.push_back(aCell);
  87
  88         ++mnCols;
  89     }
  90 };
  91
  92 }
  93
  94 namespace datastreams {
  95
  96 class ReaderThread : public salhelper::Thread
  97 {
  98     std::unique_ptr<SvStream> mpStream;
  99     size_t mnColCount;
 100     std::atomic<bool> mbTerminate;
 101
 102     std::queue<DataStream::LinesType> maPendingLines;
 103     std::queue<DataStream::LinesType> maUsedLines;
 104     std::mutex maMtxLines;
 105
 106     osl::Condition maCondReadStream;
 107     osl::Condition maCondConsume;
 108
 109     orcus::csv::parser_config maConfig;
 110
 111 public:
 112
 113     ReaderThread(std::unique_ptr<SvStream> pData, size_t nColCount):
 114         Thread("ReaderThread"),
 115         mpStream(std::move(pData)),
 116         mnColCount(nColCount),
 117         mbTerminate(false)
 118     {
 119         maConfig.delimiters.push_back(',');
 120         maConfig.text_qualifier = '"';
 121     }
 122
 123     bool isTerminateRequested()
 124     {
 125         return mbTerminate.load();
 126     }
 127
 128     void requestTerminate()
 129     {
 130         mbTerminate.store(true);
 131     }
 132
 133     void endThread()
 134     {
 135         requestTerminate();
 136         maCondReadStream.set();
 137     }
 138
 139     void waitForNewLines()
 140     {
 141         maCondConsume.wait();
 142         maCondConsume.reset();
 143     }
 144
 145     DataStream::LinesType popNewLines()
 146     {
 147         auto pLines = std::move(maPendingLines.front());
 148         maPendingLines.pop();
 149         return pLines;
 150     }
 151
 152     void resumeReadStream()
 153     {
 154         if (maPendingLines.size() <= 4)
 155             maCondReadStream.set(); // start producer again
 156     }
 157
 158     bool hasNewLines() const
 159     {
 160         return !maPendingLines.empty();
 161     }
 162
 163     void pushUsedLines( DataStream::LinesType pLines )
 164     {
 165         maUsedLines.push(std::move(pLines));
 166     }
 167
 168     std::mutex& getLinesMutex()
 169     {
 170         return maMtxLines;
 171     }
 172
 173 private:
 174     virtual void execute() override
 175     {
 176         while (!isTerminateRequested())
 177         {
 178             std::optional<DataStream::LinesType> oLines;
 179             std::unique_lock aGuard(getLinesMutex());
 180
 181             if (!maUsedLines.empty())
 182             {
 183                 // Re-use lines from previous runs.
 184                 oLines = std::move(maUsedLines.front());
 185                 maUsedLines.pop();
 186                 aGuard.unlock(); // unlock
 187             }
 188             else
 189             {
 190                 aGuard.unlock(); // unlock
 191                 oLines.emplace(10);
 192             }
 193
 194             // Read & store new lines from stream.
 195             for (DataStream::Line & rLine : *oLines)
 196             {
 197                 rLine.maCells.clear();
 198                 mpStream->ReadLine(rLine.maLine);
 199                 CSVHandler aHdl(rLine, mnColCount);
 200                 orcus::csv_parser<CSVHandler> parser(rLine.maLine, aHdl, maConfig);
 201                 parser.parse();
 202             }
 203
 204             aGuard.lock(); // lock
 205             while (!isTerminateRequested() && maPendingLines.size() >= 8)
 206             {
 207                 // pause reading for a bit
 208                 aGuard.unlock(); // unlock
 209                 maCondReadStream.wait();
 210                 maCondReadStream.reset();
 211                 aGuard.lock(); // lock
 212             }
 213             maPendingLines.push(std::move(*oLines));
 214             maCondConsume.set();
 215             if (!mpStream->good())
 216                 requestTerminate();
 217         }
 218     }
 219 };
 220
 221 }
 222
 223 DataStream::Cell::Cell() : mfValue(0.0), mbValue(true) {}
 224
 225 DataStream::Cell::Cell( const Cell& r ) : mbValue(r.mbValue)
 226 {
 227     if (r.mbValue)
 228         mfValue = r.mfValue;
 229     else
 230     {
 231         maStr.Pos = r.maStr.Pos;
 232         maStr.Size = r.maStr.Size;
 233     }
 234 }
 235
 236 void DataStream::MakeToolbarVisible()
 237 {
 238     ScViewData* pViewData = ScDocShell::GetViewData();
 239     if (!pViewData)
 240         return;
 241
 242     css::uno::Reference< css::frame::XFrame > xFrame =
 243         pViewData->GetViewShell()->GetViewFrame().GetFrame().GetFrameInterface();
 244     if (!xFrame.is())
 245         return;
 246
 247     css::uno::Reference< css::beans::XPropertySet > xPropSet(xFrame, css::uno::UNO_QUERY);
 248     if (!xPropSet.is())
 249         return;
 250
 251     css::uno::Reference< css::frame::XLayoutManager > xLayoutManager;
 252     xPropSet->getPropertyValue(u"LayoutManager"_ustr) >>= xLayoutManager;
 253     if (!xLayoutManager.is())
 254         return;
 255
 256     static constexpr OUString sResourceURL( u"private:resource/toolbar/datastreams"_ustr );
 257     css::uno::Reference< css::ui::XUIElement > xUIElement = xLayoutManager->getElement(sResourceURL);
 258     if (!xUIElement.is())
 259     {
 260         xLayoutManager->createElement( sResourceURL );
 261         xLayoutManager->showElement( sResourceURL );
 262     }
 263 }
 264
 265 DataStream* DataStream::Set(
 266     ScDocShell *pShell, const OUString& rURL, const ScRange& rRange,
 267     sal_Int32 nLimit, MoveType eMove)
 268 {
 269     DataStream* pLink = new DataStream(pShell, rURL, rRange, nLimit, eMove);
 270     sc::DocumentLinkManager& rMgr = pShell->GetDocument().GetDocLinkManager();
 271     rMgr.setDataStream(pLink);
 272     return pLink;
 273 }
 274
 275 DataStream::DataStream(ScDocShell *pShell, const OUString& rURL, const ScRange& rRange,
 276         sal_Int32 nLimit, MoveType eMove) :
 277     mpDocShell(pShell),
 278     maDocAccess(mpDocShell->GetDocument()),
 279     meOrigMove(NO_MOVE),
 280     meMove(NO_MOVE),
 281     mbRunning(false),
 282     mbValuesInLine(false),
 283     mbRefreshOnEmptyLine(false),
 284     mnLinesCount(0),
 285     mnLinesSinceRefresh(0),
 286     mfLastRefreshTime(0.0),
 287     mnCurRow(0),
 288     maImportTimer("sc DataStream maImportTimer"),
 289     mbIsFirst(true),
 290     mbIsUpdate(false)
 291 {
 292     maImportTimer.SetTimeout(0);
 293     maImportTimer.SetInvokeHandler( LINK(this, DataStream, ImportTimerHdl) );
 294
 295     Decode(rURL, rRange, nLimit, eMove);
 296 }
 297
 298 DataStream::~DataStream()
 299 {
 300     if (mbRunning)
 301         StopImport();
 302
 303     if (mxReaderThread.is())
 304     {
 305         mxReaderThread->endThread();
 306         mxReaderThread->join();
 307     }
 308     moLines.reset();
 309 }
 310
 311 DataStream::Line DataStream::ConsumeLine()
 312 {
 313     if (!moLines || mnLinesCount >= moLines->size())
 314     {
 315         mnLinesCount = 0;
 316         if (mxReaderThread->isTerminateRequested())
 317             return Line();
 318
 319         std::unique_lock aGuard(mxReaderThread->getLinesMutex());
 320         if (moLines)
 321         {
 322             mxReaderThread->pushUsedLines(std::move(*moLines));
 323             moLines.reset();
 324         }
 325
 326         while (!mxReaderThread->hasNewLines())
 327         {
 328             aGuard.unlock(); // unlock
 329             mxReaderThread->waitForNewLines();
 330             aGuard.lock(); // lock
 331         }
 332
 333         moLines = mxReaderThread->popNewLines();
 334         mxReaderThread->resumeReadStream();
 335     }
 336     return moLines->at(mnLinesCount++);
 337 }
 338
 339 ScRange DataStream::GetRange() const
 340 {
 341     ScRange aRange = maStartRange;
 342     aRange.aEnd = maEndRange.aEnd;
 343     return aRange;
 344 }
 345
 346 void DataStream::Decode(const OUString& rURL, const ScRange& rRange,
 347         sal_Int32 nLimit, MoveType eMove)
 348 {
 349     msURL = rURL;
 350     meMove = eMove;
 351     meOrigMove = eMove;
 352
 353     mbValuesInLine = true; // always true.
 354
 355     mnCurRow = rRange.aStart.Row();
 356
 357     ScRange aRange = rRange;
 358     if (aRange.aStart.Row() != aRange.aEnd.Row())
 359         // We only allow this range to be one row tall.
 360         aRange.aEnd.SetRow(aRange.aStart.Row());
 361
 362     maStartRange = aRange;
 363     maEndRange = aRange;
 364     const auto & rDoc = mpDocShell->GetDocument();
 365     if (nLimit == 0)
 366     {
 367         // Unlimited
 368         maEndRange.aStart.SetRow(rDoc.MaxRow());
 369     }
 370     else if (nLimit > 0)
 371     {
 372         // Limited.
 373         maEndRange.aStart.IncRow(nLimit-1);
 374         if (maEndRange.aStart.Row() > rDoc.MaxRow())
 375             maEndRange.aStart.SetRow(rDoc.MaxRow());
 376     }
 377
 378     maEndRange.aEnd.SetRow(maEndRange.aStart.Row());
 379 }
 380
 381 void DataStream::StartImport()
 382 {
 383     if (mbRunning)
 384         return;
 385
 386     if (!mxReaderThread.is())
 387     {
 388         std::unique_ptr<SvStream> pStream(new SvFileStream(msURL, StreamMode::READ));
 389         mxReaderThread = new datastreams::ReaderThread(std::move(pStream), maStartRange.aEnd.Col() - maStartRange.aStart.Col() + 1);
 390         mxReaderThread->launch();
 391     }
 392     mbRunning = true;
 393     maDocAccess.reset();
 394
 395     maImportTimer.Start();
 396 }
 397
 398 void DataStream::StopImport()
 399 {
 400     if (!mbRunning)
 401         return;
 402
 403     mbRunning = false;
 404     Refresh();
 405     maImportTimer.Stop();
 406 }
 407
 408 void DataStream::SetRefreshOnEmptyLine( bool bVal )
 409 {
 410     mbRefreshOnEmptyLine = bVal;
 411 }
 412
 413 void DataStream::Refresh()
 414 {
 415     Application::Yield();
 416
 417     double fStart = getNow();
 418
 419     // Hard recalc will repaint the grid area.
 420     mpDocShell->DoHardRecalc();
 421     mpDocShell->SetDocumentModified();
 422
 423     fTimes[ DebugTime::Recalc ] = getNow() - fStart;
 424
 425     mfLastRefreshTime = getNow();
 426     mnLinesSinceRefresh = 0;
 427 }
 428
 429 void DataStream::MoveData()
 430 {
 431     switch (meMove)
 432     {
 433         case RANGE_DOWN:
 434         {
 435             if (mnCurRow == maEndRange.aStart.Row())
 436                 meMove = MOVE_UP;
 437         }
 438         break;
 439         case MOVE_UP:
 440         {
 441             mbIsUpdate = true;
 442             // Remove the top row and shift the remaining rows upward. Then
 443             // insert a new row at the end row position.
 444             ScRange aRange = maStartRange;
 445             aRange.aEnd = maEndRange.aEnd;
 446             maDocAccess.shiftRangeUp(aRange);
 447         }
 448         break;
 449         case MOVE_DOWN:
 450         {
 451             mbIsUpdate = true;
 452             // Remove the end row and shift the remaining rows downward by
 453             // inserting a new row at the top row.
 454             ScRange aRange = maStartRange;
 455             aRange.aEnd = maEndRange.aEnd;
 456             maDocAccess.shiftRangeDown(aRange);
 457         }
 458         break;
 459         case NO_MOVE:
 460         default:
 461             ;
 462     }
 463     if(mbIsFirst && mbIsUpdate)
 464     {
 465         sal_Int32 nStreamTimeout = officecfg::Office::Calc::DataStream::UpdateTimeout::get();
 466         maImportTimer.SetTimeout(nStreamTimeout);
 467         mbIsFirst = false;
 468     }
 469 }
 470
 471 void DataStream::Text2Doc()
 472 {
 473     Line aLine = ConsumeLine();
 474     if (aLine.maCells.empty() && mbRefreshOnEmptyLine)
 475     {
 476         // Empty line detected.  Trigger refresh and discard it.
 477         Refresh();
 478         return;
 479     }
 480
 481     double fStart = getNow();
 482
 483     MoveData();
 484     {
 485         SCCOL nCol = maStartRange.aStart.Col();
 486         const char* pLineHead = aLine.maLine.getStr();
 487         for (const Cell& rCell : aLine.maCells)
 488         {
 489             if (rCell.mbValue)
 490             {
 491                 maDocAccess.setNumericCell(
 492                     ScAddress(nCol, mnCurRow, maStartRange.aStart.Tab()), rCell.mfValue);
 493             }
 494             else
 495             {
 496                 maDocAccess.setStringCell(
 497                     ScAddress(nCol, mnCurRow, maStartRange.aStart.Tab()),
 498                     OUString(pLineHead+rCell.maStr.Pos, rCell.maStr.Size, RTL_TEXTENCODING_UTF8));
 499             }
 500             ++nCol;
 501         }
 502     }
 503
 504     fTimes[ DebugTime::Import ] = getNow() - fStart;
 505
 506     if (meMove == NO_MOVE)
 507         return;
 508
 509     if (meMove == RANGE_DOWN)
 510     {
 511         ++mnCurRow;
 512 //      mpDocShell->GetViewData().GetView()->AlignToCursor(
 513 //              maStartRange.aStart.Col(), mnCurRow, SC_FOLLOW_JUMP);
 514     }
 515
 516     if (getNow() - mfLastRefreshTime > 0.1 && mnLinesSinceRefresh > 200)
 517         // Refresh no more frequently than every 0.1 second, and wait until at
 518         // least we have processed 200 lines.
 519         Refresh();
 520
 521     ++mnLinesSinceRefresh;
 522 }
 523
 524 bool DataStream::ImportData()
 525 {
 526     if (!mbValuesInLine)
 527         // We no longer support this mode. To be deleted later.
 528         return false;
 529
 530     ScViewData* pViewData = ScDocShell::GetViewData();
 531     if (!pViewData)
 532         return false;
 533
 534     if (pViewData->GetViewShell()->NeedsRepaint())
 535         return mbRunning;
 536
 537     Text2Doc();
 538     return mbRunning;
 539 }
 540
 541 IMPL_LINK_NOARG(DataStream, ImportTimerHdl, Timer *, void)
 542 {
 543     if (ImportData())
 544         maImportTimer.Start();
 545 }
 546
 547 } // namespace sc
 548
 549 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */