ossim::S3StreamBuffer Class Reference

#include <ossimS3StreamBuffer.h>

Inheritance diagram for ossim::S3StreamBuffer:

Classes
class	blockInfo

Public Member Functions
	S3StreamBuffer (ossim_int64 blockSize=ossim::S3StreamDefaults::m_readBlocksize)
S3StreamBuffer *	open (const char *connectionString, const ossimKeywordlist &options, std::ios_base::openmode mode)
S3StreamBuffer *	open (const std::string &connectionString, const ossimKeywordlist &options, std::ios_base::openmode mode)
bool	is_open () const
virtual	~S3StreamBuffer ()
ossim_uint64	getFileSize () const
ossim_uint64	getBlockSize () const

Protected Member Functions
virtual pos_type	seekoff (off_type offset, std::ios_base::seekdir dir, std::ios_base::openmode __mode=std::ios_base::in|std::ios_base::out)
virtual pos_type	seekpos (pos_type pos, std::ios_base::openmode mode=std::ios_base::in|std::ios_base::out)
virtual std::streamsize	xsgetn (char_type *s, std::streamsize n)
virtual int	underflow ()
void	clearAll ()
ossim_int64	getBlockIndex (ossim_int64 byteOffset) const
ossim_int64	getBlockOffset (ossim_int64 byteOffset) const
bool	getBlockRangeInBytes (ossim_int64 blockIndex, ossim_int64 &startRange, ossim_int64 &endRange) const
bool	loadBlock (ossim_int64 absolutePosition)
ossim_int64	getAbsoluteByteOffset () const
bool	withinWindow () const

Protected Attributes
std::shared_ptr< Aws::S3::S3Client >	m_client
std::string	m_bucket
std::string	m_key
std::vector< char >	m_buffer
ossim_int64	m_bufferActualDataSize
ossim_int64	m_currentBlockPosition
char *	m_bufferPtr
ossim_int64	m_fileSize
bool	m_opened
blockInfo	m_blockInfo

Detailed Description

Definition at line 35 of file ossimS3StreamBuffer.h.

Constructor & Destructor Documentation

S3StreamBuffer()

ossim::S3StreamBuffer::S3StreamBuffer

(

ossim_int64

blockSize = ossim::S3StreamDefaults::m_readBlocksize

)

Definition at line 50 of file ossimS3StreamBuffer.cpp.

References ossim::AwsStreamFactory::getSharedS3Client(), ossim::AwsStreamFactory::instance(), ossimPreferences::instance(), m_bufferPtr, and m_client.

   :
   m_bucket(""),
   m_key(""),
   m_buffer(blockSize),
   m_bufferActualDataSize(0),
   m_currentBlockPosition(-1),
   m_bufferPtr(0),
   m_fileSize(0),
   m_opened(false)
   //m_mode(0)
{
   Aws::Client::ClientConfiguration config;

   // Look for AWS S3 regionn override:
   std::string region = ossimPreferences::instance()->
      preferencesKWL().findKey(std::string("ossim.plugins.aws.s3.region"));
   if ( region.size() )
   {
      config.region = region.c_str();
   }
   m_client = ossim::AwsStreamFactory::instance()->getSharedS3Client();
  // m_client = new Aws::S3::S3Client( config );
   
//    std::cout << "CONSTRUCTED!!!!!" << std::endl;
//  setp(0);
   setg(m_bufferPtr, m_bufferPtr, m_bufferPtr);
}

~S3StreamBuffer()

ossim::S3StreamBuffer::~S3StreamBuffer ( )

virtual

Definition at line 79 of file ossimS3StreamBuffer.cpp.

{
}

Member Function Documentation

clearAll()

void ossim::S3StreamBuffer::clearAll ( )

protected

Definition at line 268 of file ossimS3StreamBuffer.cpp.

{
   m_bucket = "";
   m_key    = "";
   m_fileSize = 0;
   m_opened = false;
   m_currentBlockPosition = 0;
   m_blockInfo.setBytes(0,0,0);
}

getAbsoluteByteOffset()

ossim_int64 ossim::S3StreamBuffer::getAbsoluteByteOffset ( ) const

protected

Definition at line 541 of file ossimS3StreamBuffer.cpp.

{
   ossim_int64 result = -1;
   
   if(m_currentBlockPosition >= 0)
   {
      result = m_blockInfo.getCurrentByte();
      //result = m_currentBlockPosition;
      // if(gptr()&&eback())
      // {
      //    result += (gptr()-eback());
      // }
   }

   // std::cout << "RESULT getAbsoluteByteOffset======== " << result << "\n";
   return result;
}

getBlockIndex()

ossim_int64 ossim::S3StreamBuffer::getBlockIndex ( ossim_int64  byteOffset ) const

protected

Definition at line 83 of file ossimS3StreamBuffer.cpp.

{
   ossim_int64 blockNumber = -1;
  
   if(byteOffset < (ossim_int64)m_fileSize)
   {
      if(m_buffer.size()>0)
      {
         blockNumber = byteOffset/m_buffer.size();
      }    
   }

   return blockNumber;
}

getBlockOffset()

ossim_int64 ossim::S3StreamBuffer::getBlockOffset ( ossim_int64  byteOffset ) const

protected

Definition at line 98 of file ossimS3StreamBuffer.cpp.

{
   ossim_int64 blockOffset = -1;
  
   if(m_buffer.size()>0)
   {
      blockOffset = byteOffset%m_buffer.size();
   }    

   return blockOffset;
}

getBlockRangeInBytes()

bool ossim::S3StreamBuffer::getBlockRangeInBytes ( ossim_int64  blockIndex, ossim_int64 &  startRange, ossim_int64 &  endRange  ) const

protected

Definition at line 110 of file ossimS3StreamBuffer.cpp.

{
   bool result = false;

   if(blockIndex >= 0)
   {
      startRange = blockIndex*m_buffer.size();
      endRange = startRange + m_buffer.size()-1;

      result = true;    
   }

   return result;
}

getBlockSize()

ossim_uint64 ossim::S3StreamBuffer::getBlockSize

(

)

const

Returns

Size of block buffer in bytes.

Definition at line 570 of file ossimS3StreamBuffer.cpp.

Referenced by ossim::S3IStream::getBlockSize().

{
   return m_buffer.size();
}

getFileSize()

ossim_uint64 ossim::S3StreamBuffer::getFileSize

(

)

const

Returns

Size of file in bytes.

Definition at line 565 of file ossimS3StreamBuffer.cpp.

Referenced by ossim::S3IStream::getFileSize().

{
   return static_cast<ossim_uint32>(m_fileSize);
}

is_open()

bool ossim::S3StreamBuffer::is_open ( ) const

inline

Definition at line 78 of file ossimS3StreamBuffer.h.

References m_opened.

   {
      return m_opened;
   }

loadBlock()

bool ossim::S3StreamBuffer::loadBlock ( ossim_int64  absolutePosition )

protected

Definition at line 127 of file ossimS3StreamBuffer.cpp.

{
   if(traceDebug())
   {
      ossimNotify(ossimNotifyLevel_DEBUG)
         << "ossim::S3StreamBuffer::loadBlock DEBUG: entered with absolute position: " << absolutePosition << "\n";

   }
   bool result = false;
   m_bufferPtr = 0;
   GetObjectRequest getObjectRequest;
   std::stringstream stringStream;
   ossim_int64 startRange, endRange;
   ossim_int64 blockIndex = getBlockIndex(absolutePosition);
   if((absolutePosition < 0) || (absolutePosition > (ossim_int64)m_fileSize)) return false;
   //std::cout << "CURRENT BYTE LOCATION = " << absoluteLocation << std::endl;
   if(getBlockRangeInBytes(blockIndex, startRange, endRange))
   {
      stringStream << "bytes=" << startRange << "-" << endRange;
      getObjectRequest.WithBucket(m_bucket.c_str())
         .WithKey(m_key.c_str()).WithRange(stringStream.str().c_str());
      auto getObjectOutcome = m_client->GetObject(getObjectRequest);

      if(getObjectOutcome.IsSuccess())
      {
//        std::cout << "GOOD CALL!!!!!!!!!!!!\n";
         Aws::IOStream& bodyStream = getObjectOutcome.GetResult().GetBody();
         ossim_int64 bufSize = getObjectOutcome.GetResult().GetContentLength();
//        std::cout << "SIZE OF RESULT ======== " << bufSize << std::endl;
         m_bufferActualDataSize = bufSize;
         bodyStream.read(&m_buffer.front(), bufSize);
         m_bufferPtr = &m_buffer.front();

         ossim_int64 delta = absolutePosition-startRange;
         setg(m_bufferPtr, m_bufferPtr + delta, m_bufferPtr+m_bufferActualDataSize);
         m_blockInfo.setBytes(startRange,startRange+delta,startRange+m_bufferActualDataSize);
         // std::cout << "LOADING BLOCK: " << m_blockInfo.getStartByte() << ", " << m_blockInfo.getCurrentByte() << ", " << m_blockInfo.getEndByte() << "\n";
         m_currentBlockPosition = startRange;
         result = true;
         //std::cout << "Successfully retrieved object from s3 with value: " << std::endl;
         //std::cout << getObjectOutcome.GetResult().GetBody().rdbuf() << std::endl << std::endl;;  
      }
      else
      {
         m_bufferActualDataSize = 0;
      }
   }
   if(traceDebug())
   {
      ossimNotify(ossimNotifyLevel_DEBUG)
         << "ossim::S3StreamBuffer::loadBlock DEBUG: leaving with absolutePosition " << absolutePosition << "\n";

   }

   return result;
}

open() [1/2]

ossim::S3StreamBuffer * ossim::S3StreamBuffer::open	(	const char *	connectionString,
		const ossimKeywordlist &	options,
		std::ios_base::openmode	mode
	)

Definition at line 184 of file ossimS3StreamBuffer.cpp.

Referenced by ossim::S3IStream::open().

{
   std::string temp(connectionString);
   return open(temp, options, m);
}

open() [2/2]

ossim::S3StreamBuffer * ossim::S3StreamBuffer::open	(	const std::string &	connectionString,
		const ossimKeywordlist &	options,
		std::ios_base::openmode	mode
	)

Definition at line 192 of file ossimS3StreamBuffer.cpp.

{
   if(traceDebug())
   {
      ossimNotify(ossimNotifyLevel_DEBUG)
         << "ossim::S3StreamBuffer::open DEBUG: entered..... with connection " << connectionString << std::endl;
   }
   // bool result = false;
   ossimUrl url(connectionString);
   clearAll();
   // m_mode = mode;
   ossimTimer::Timer_t startTimer = ossimTimer::instance()->tick();

   // AWS server is case insensitive:
   if( (url.getProtocol() == "s3") || (url.getProtocol() == "S3") )
   {
      ossim_int64 filesize;
      m_bucket = url.getIp().c_str();
      m_key = url.getPath().c_str();
      if(ossim::S3HeaderCache::instance()->getCachedFilesize(connectionString, filesize))
      {
         m_fileSize = filesize;
         if(m_fileSize >= 0)
         {
            m_opened = true;
            m_currentBlockPosition = 0;
         }
      }
      else
      {
         if(!m_bucket.empty() && !m_key.empty())
         {
            HeadObjectRequest headObjectRequest;
            headObjectRequest.WithBucket(m_bucket.c_str())
               .WithKey(m_key.c_str());
            auto headObject = m_client->HeadObject(headObjectRequest);
            if(headObject.IsSuccess())
            {
               m_fileSize = headObject.GetResult().GetContentLength();
               m_opened = true;
               m_currentBlockPosition = 0;
               ossim::S3HeaderCache::Node_t nodePtr = std::make_shared<ossim::S3HeaderCacheNode>(m_fileSize);
               ossim::S3HeaderCache::instance()->addHeader(connectionString, nodePtr);
            }
            else
            {
               m_opened = false;
               m_fileSize = -1;
               m_currentBlockPosition = 0;
               ossim::S3HeaderCache::Node_t nodePtr = std::make_shared<ossim::S3HeaderCacheNode>(m_fileSize);
               ossim::S3HeaderCache::instance()->addHeader(connectionString, nodePtr);               
            }
         }
      }
   }
  ossimTimer::Timer_t endTimer = ossimTimer::instance()->tick();

   if(traceDebug())
   {
      ossim_float64 delta = ossimTimer::instance()->delta_s(startTimer, endTimer);

      ossimNotify(ossimNotifyLevel_DEBUG)
         << "ossim::S3StreamBuffer::open DEBUG: Took " << delta << " seconds to open" << std::endl;
      ossimNotify(ossimNotifyLevel_DEBUG)
         << "ossim::S3StreamBuffer::open DEBUG: leaving....." << std::endl;

   }
   if(m_opened) return this;

   return 0;
}

seekoff()

ossim::S3StreamBuffer::pos_type ossim::S3StreamBuffer::seekoff ( off_type  offset, std::ios_base::seekdir  dir, std::ios_base::openmode  __mode = std::ios_base::in | std::ios_base::out  )

protectedvirtual

Definition at line 309 of file ossimS3StreamBuffer.cpp.

{ 
   // std::cout <<"ossim::S3StreamBuffer::seekoff type size === " << sizeof(off_type) << std::endl;
   // std::cout <<"ossim::S3StreamBuffer::seekoff offset ====== " << offset << std::endl;
  // std::cout << "ossim::S3StreamBuffer::seekoff\n";
   pos_type result = pos_type(off_type(-1));
   // bool withinBlock = true;
   if((mode & std::ios_base::in)&&
      (mode & std::ios_base::out))
   {
      return result;
   }
   switch(dir)
   {
      case std::ios_base::beg:
      {
         ossim_int64 absolutePosition = getAbsoluteByteOffset();
         // really would like to figure out a better way but for now
         // we have to have one valid block read in to properly adjust
         // gptr() with gbump
         //
         // if(!gptr())
         // {
         //    if(!loadBlock(offset))
         //    {
         //       return result;
         //    }
         // }
         if((offset <= (ossim_int64)m_fileSize)&&
            (offset >=0))
         {
            result = pos_type(offset);
         }
         // if(!gptr())
         // {
         //    if(!loadBlock(result))
         //    {
         //       return EOF;
         //    }
         // }
         //else 
         if(mode  & std::ios_base::in)
         {
            absolutePosition = getAbsoluteByteOffset();
            ossim_int64 delta = offset - absolutePosition;
            setg(eback(), gptr()+delta, egptr());
            m_blockInfo.setCurrentByte(offset);
         }
         // std::cout << "ossim::S3StreamBuffer::seekoff beg RESULT??????????????????? " << result << std::endl;
         break;
      }
      case std::ios_base::cur:
      {
         // if(!gptr())
         // {
         //    // std::cout << "LOADING BLOCK!!!!!!!!!!!!!!\n" << std::endl;
         //    if(!loadBlock(0))
         //    {
         //       return result;
         //    }
         // }
         // result = getAbsoluteByteOffset();
         // std::cout << "INITIAL ABSOLUTE BYTE OFFSET ==== " << result << "\n";
         if(!offset)
         {
            result = getAbsoluteByteOffset();
         }
         else
         {
            result += offset;
            setg(eback(), gptr()+offset, egptr());
            m_blockInfo.setCurrentByte(m_blockInfo.getCurrentByte()+offset);
         }

        break;
      }
      case std::ios_base::end:
      {
         ossim_int64 absolutePosition  = m_fileSize + offset;
         // if(!gptr())
         // {
         //    if(!loadBlock(absolutePosition))
         //    {
         //       return result;
         //    }               
         // }
         ossim_int64 currentAbsolutePosition = getAbsoluteByteOffset();
         ossim_int64 delta = absolutePosition-currentAbsolutePosition;

//         std::cout << "CURRENT ABSOLUTE POSITION === " << currentAbsolutePosition << std::endl;
//         std::cout << "CURRENT ABSOLUTE delta POSITION === " << delta << std::endl;
         if(mode & std::ios_base::in )
         {
            setg(eback(), gptr()+delta, egptr());
            m_blockInfo.setCurrentByte(absolutePosition);
            result = absolutePosition;
         }
         // std::cout << "ossim::S3StreamBuffer::seekoff end RESULT??????????????????? " << result << std::endl;
         break;
      }
      default:
      {
         break;
      }
   }

   return result; 
} 

seekpos()

ossim::S3StreamBuffer::pos_type ossim::S3StreamBuffer::seekpos ( pos_type  pos, std::ios_base::openmode  mode = std::ios_base::in | std::ios_base::out  )

protectedvirtual

Definition at line 420 of file ossimS3StreamBuffer.cpp.

{
   // std::cout << "ossim::S3StreamBuffer::seekpos: " << pos << std::endl;
   pos_type result = pos_type(off_type(-1));
   ossim_int64 tempPos = static_cast<ossim_int64>(pos);
   // Currently we must initialize to a block
   // if(!gptr())
   // {
   //    if(!loadBlock(tempPos))
   //    {
   //       return result;
   //    }
   // }
   ossim_int64 absoluteLocation = getAbsoluteByteOffset();
   if(mode & std::ios_base::in)
   {
      if((pos >= 0)&&(pos < (ossim_int64)m_fileSize))
      {
         m_blockInfo.setCurrentByte(tempPos);
         ossim_int64 delta = tempPos-absoluteLocation;
         if(delta)
         {
            setg(eback(), gptr()+delta, egptr());
         }
         result = pos;
      }
   }
   return result;
}

underflow()

int ossim::S3StreamBuffer::underflow ( )

protectedvirtual

Definition at line 279 of file ossimS3StreamBuffer.cpp.

{
   if(!is_open())
   {
      return EOF;
   }
   else if( !m_blockInfo.withinWindow() )
   {
      ossim_int64 absolutePosition = getAbsoluteByteOffset();
      if(absolutePosition < 0)
      {
         return EOF;
      }

      if(!loadBlock(absolutePosition))
      {
         return EOF;
      }
   }
   
   // std::cout << "GPTR CHARACTER ========== "
   // << (int)static_cast<ossim_uint8>(*gptr()) << std::endl;

   //---
   // Double cast to get non-negative values so as to not send an inadvertent
   // EOF(-1) to caller.
   //---
   return (int)static_cast<ossim_uint8>(*gptr());
}

withinWindow()

bool ossim::S3StreamBuffer::withinWindow ( ) const

protected

Definition at line 559 of file ossimS3StreamBuffer.cpp.

{
   if(!gptr()) return false;
   return ((gptr()>=eback()) && (gptr()<egptr()));
}

xsgetn()

std::streamsize ossim::S3StreamBuffer::xsgetn ( char_type *  s, std::streamsize  n  )

protectedvirtual

Definition at line 450 of file ossimS3StreamBuffer.cpp.

References n.

{      
  // std::cout << "ossim::S3StreamBuffer::xsgetn" << std::endl;

   if(!is_open()) return EOF;
   // unsigned long int bytesLeftToRead = egptr()-gptr();
   // initialize if we need to to load the block at current position
//   if((!withinWindow())&&is_open())
   if((!m_blockInfo.withinWindow())&&is_open())
   {
      // if(!gptr())
      // {
      //    if(!loadBlock(0))
      //    {
      //       return EOF;
      //    }
      // }
      // else if(!loadBlock(getAbsoluteByteOffset()))
      // {
      //    return EOF;
      // }
      if(!loadBlock(m_blockInfo.getCurrentByte()))
      {
         return EOF;
      }
   }
   ossim_int64 bytesNeedToRead = n;
   // ossim_int64 bytesToRead = 0;
   ossim_int64 bytesRead = 0;
   ossim_int64 currentAbsolutePosition = m_blockInfo.getCurrentByte();//getAbsoluteByteOffset();
   // ossim_int64 startOffset, endOffset;
   if(currentAbsolutePosition >= (ossim_int64)m_fileSize)
   {
      return EOF;
   }
   else if((currentAbsolutePosition + bytesNeedToRead)>(ossim_int64)m_fileSize)
   {
      bytesNeedToRead = (m_fileSize - currentAbsolutePosition);
   }

   while(bytesNeedToRead > 0)
   {
      currentAbsolutePosition = m_blockInfo.getCurrentByte();//getAbsoluteByteOffset();
      //if(!withinWindow())
      if(!m_blockInfo.withinWindow())
      {
         if(!loadBlock(m_blockInfo.getCurrentByte()))
         {
            return bytesRead;
         }
         currentAbsolutePosition = m_blockInfo.getCurrentByte();
      }

      // get each bloc  
      if(currentAbsolutePosition>=0)
      {
         //getBlockRangeInBytes(getBlockIndex(m_currentBlockPosition), startOffset, endOffset);      
    
         ossim_int64 delta = (m_blockInfo.getEndByte()-m_blockInfo.getCurrentByte());//(endOffset - m_currentBlockPosition)+1;

         if(delta <= bytesNeedToRead)
         {

            std::memcpy(s+bytesRead, gptr(), delta);
            //m_currentBlockPosition += delta;
            //setg(eback(), egptr(), egptr());
            bytesRead+=delta;
            bytesNeedToRead-=delta;
            //gbump(delta);
            setg(eback(), gptr()+delta, egptr());
            m_blockInfo.setCurrentByte(m_blockInfo.getCurrentByte()+delta);
         }
         else
         {
            std::memcpy(s+bytesRead, gptr(), bytesNeedToRead);
            setg(eback(), gptr()+bytesNeedToRead, egptr());
            m_blockInfo.setCurrentByte(m_blockInfo.getCurrentByte()+bytesNeedToRead);
            //gbump(bytesNeedToRead);
//            std::cout << "gbump 2 DELTA ========= " << bytesNeedToRead << std::endl;
            bytesRead+=bytesNeedToRead;
            bytesNeedToRead=0;
         }
      }
      else
      {
         break;
      }
   }
   return std::streamsize(bytesRead);
}

Member Data Documentation

m_blockInfo

blockInfo ossim::S3StreamBuffer::m_blockInfo

protected

Definition at line 128 of file ossimS3StreamBuffer.h.

m_bucket

std::string ossim::S3StreamBuffer::m_bucket

protected

Definition at line 120 of file ossimS3StreamBuffer.h.

m_buffer

std::vector<char> ossim::S3StreamBuffer::m_buffer

protected

Definition at line 122 of file ossimS3StreamBuffer.h.

m_bufferActualDataSize

ossim_int64 ossim::S3StreamBuffer::m_bufferActualDataSize

protected

Definition at line 123 of file ossimS3StreamBuffer.h.

m_bufferPtr

char* ossim::S3StreamBuffer::m_bufferPtr

protected

Definition at line 125 of file ossimS3StreamBuffer.h.

Referenced by S3StreamBuffer().

m_client

std::shared_ptr<Aws::S3::S3Client> ossim::S3StreamBuffer::m_client

mutableprotected

Definition at line 119 of file ossimS3StreamBuffer.h.

Referenced by S3StreamBuffer().

m_currentBlockPosition

ossim_int64 ossim::S3StreamBuffer::m_currentBlockPosition

protected

Definition at line 124 of file ossimS3StreamBuffer.h.

m_fileSize

ossim_int64 ossim::S3StreamBuffer::m_fileSize

protected

Definition at line 126 of file ossimS3StreamBuffer.h.

m_key

std::string ossim::S3StreamBuffer::m_key

protected

Definition at line 121 of file ossimS3StreamBuffer.h.

m_opened

bool ossim::S3StreamBuffer::m_opened

protected

Definition at line 127 of file ossimS3StreamBuffer.h.

Referenced by is_open().

---

The documentation for this class was generated from the following files:

• ossimS3StreamBuffer.h
• ossimS3StreamBuffer.cpp