PublishStore.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2025 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//

#ifndef _PUBLISHSTORE_H_
#define _PUBLISHSTORE_H_

#include <amps/ampsplusplus.hpp>
#include <amps/BlockPublishStore.hpp>
#include <amps/MMapStoreBuffer.hpp>
#ifndef _WIN32
  #include <unistd.h>
#endif


namespace AMPS
{
  class PublishStore : public BlockPublishStore
  {
  public:
    PublishStore(const std::string& fileName_, bool errorOnPublishGap_ = false)
      : BlockPublishStore(new MMapStoreBuffer(fileName_), 1000,
                          true, errorOnPublishGap_)
      , _fileName(fileName_)
      , _initialBlocks(1000)
      , _truncateOnClose(false)
    {
      recover();
    }

    PublishStore(const std::string& fileName_, size_t blocksPerRealloc_,
                 bool errorOnPublishGap_ = false)
      : BlockPublishStore(new MMapStoreBuffer(fileName_, blocksPerRealloc_ * 2048),
                          (amps_uint32_t)blocksPerRealloc_,
                          true, errorOnPublishGap_)
      , _fileName(fileName_)
      , _initialBlocks((int)blocksPerRealloc_)
      , _truncateOnClose(false)
    {
      recover();
    }

    PublishStore(const std::string& fileName_, size_t blocksPerRealloc_,
                 amps_uint32_t blockSize_, bool errorOnPublishGap_ = false)
      : BlockPublishStore(new MMapStoreBuffer(fileName_, blocksPerRealloc_ * blockSize_),
                          (amps_uint32_t)blocksPerRealloc_,
                          true, errorOnPublishGap_, blockSize_)
      , _fileName(fileName_)
      , _initialBlocks((int)blocksPerRealloc_)
      , _truncateOnClose(false)
    {
      recover();
    }

    ~PublishStore()
    {
      close();
    }

    void truncateOnClose(bool truncate_)
    {
      _truncateOnClose = truncate_;
    }

    void close()
    {
      if (!_blockStore.getBuffer())
      {
        return;
      }
      amps_uint64_t unpersisted = unpersistedCount();
      BufferLock guard(_blockStore);
      reinterpret_cast<MMapStoreBuffer*>(_blockStore.getBuffer())->close();
      if (_truncateOnClose && unpersisted == 0)
      {
#ifdef _WIN32
        size_t retries = 0;
        HANDLE file = INVALID_HANDLE_VALUE;
        while ( file == INVALID_HANDLE_VALUE && retries++ < 5)
        {
          file = CreateFileA(_fileName.c_str(),
                             GENERIC_READ | GENERIC_WRITE, 0, NULL,
                             OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
        }
        // Ignore failure, since this could be in destructor
        if ( file != INVALID_HANDLE_VALUE )
        {
          SetFilePointer(file, (LONG)(_initialBlocks * _blockStore.getBlockSize()),
                         NULL, FILE_BEGIN);
          SetEndOfFile(file);
          CloseHandle(file);
        }
#else
        if (truncate(_fileName.c_str(),
                     (off_t)_initialBlocks * (off_t)_blockStore.getBlockSize()) == -1)
        {
          // Ignore failure, since this could be in destructor
          ;
        }
#endif
      }
    }

    void sync()
    {
      BufferLock guard(_blockStore);
      reinterpret_cast<MMapStoreBuffer*>(_blockStore.getBuffer())->sync();
    }
  private:
    std::string _fileName;
    int _initialBlocks;
    bool _truncateOnClose;
  };

} //namespace AMPS

#endif //_PUBLISHSTORE_H_