PublishStore.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2026 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_, AMPS_MEMORYBUFFER_DEFAULT_LENGTH,
                              DEFAULT_BLOCK_SIZE, true), 1000,
                          true, errorOnPublishGap_)
      , _fileName(fileName_)
      , _initialBlocks(1000)
      , _truncateOnClose(false)
    {
      recoverUsingPersistedBlockSize();
    }

    PublishStore(const std::string& fileName_, size_t blocksPerRealloc_,
                 bool errorOnPublishGap_ = false)
      : BlockPublishStore(new MMapStoreBuffer(
                              fileName_,
                              getInitialBufferSize(blocksPerRealloc_, DEFAULT_BLOCK_SIZE),
                              normalizeBlockSize(
                                  blocksPerRealloc_, DEFAULT_BLOCK_SIZE), true),
                          normalizeBlocksPerRealloc(
                              blocksPerRealloc_, DEFAULT_BLOCK_SIZE),
                          true, errorOnPublishGap_,
                          normalizeBlockSize(
                              blocksPerRealloc_, DEFAULT_BLOCK_SIZE))
      , _fileName(fileName_)
      , _initialBlocks((int)normalizeBlocksPerRealloc(
                            blocksPerRealloc_, DEFAULT_BLOCK_SIZE))
      , _truncateOnClose(false)
    {
      recoverUsingPersistedBlockSize();
    }

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

    ~PublishStore()
    {
      close(); // -V1053 safe for our classes
    }

    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:
    void recoverUsingPersistedBlockSize()
    {
      {
        BufferLock guard(_blockStore);
        MMapStoreBuffer* buffer = reinterpret_cast<MMapStoreBuffer*>(
                                    _blockStore.getBuffer());
        if (buffer->getSize() >= sizeof(amps_uint32_t))
        {
          buffer->setPosition(0);
          amps_uint32_t persistedBlockSize = buffer->getUint32();
          // Zero indicates a newly-created store. One indicates the old
          // publish-store format, which recovers using the configured size.
          if (persistedBlockSize > 1)
          {
            _blockStore.setBlockSize(persistedBlockSize);
            buffer->setSizeAlignment(persistedBlockSize);
          }
        }
      }
      BlockPublishStore::recover();
    }

    std::string _fileName;
    int _initialBlocks;
    bool _truncateOnClose;
  };

} //namespace AMPS

#endif //_PUBLISHSTORE_H_