public long append ( Entry entry ) { Assert . notNull ( entry , "entry" ) ; Assert . stateNot ( isFull ( ) , "segment is full" ) ; long index = nextIndex ( ) ; Assert . index ( index == entry . getIndex ( ) , "inconsistent index: %s" , entry . getIndex ( ) ) ; // Calculate the offset of the entry. long offset = relativeOffset ( index ) ; // Get the term from the entry. long term = entry . getTerm ( ) ; // Get the highest term in the index. long lastTerm = termIndex . term ( ) ; // The entry term must be positive and >= the last term in the segment. Assert . arg ( term > 0 && term >= lastTerm , "term must be monotonically increasing" ) ; // Mark the starting position of the record and record the starting position of the new entry. long position = buffer . position ( ) ; // Determine whether to skip writing the term to the segment. boolean skipTerm = term == lastTerm ; // Calculate the length of the entry header bytes. int headerLength = INTEGER + LONG + BOOLEAN + ( skipTerm ? 0 : LONG ) ; // Clear the memory and skip the size and header. memory . clear ( ) . skip ( headerLength ) ; // Serialize the object into the in-memory buffer. serializer . writeObject ( entry , memory ) ; // Flip the in-memory buffer indexes. memory . flip ( ) ; // The total length of the entry is the in-memory buffer limit. int totalLength = ( int ) memory . limit ( ) ; // Calculate the length of the serialized bytes based on the in-memory buffer limit and header length. int entryLength = totalLength - headerLength ; // Set the entry size. entry . setSize ( totalLength ) ; // Compute the checksum for the entry. Checksum crc32 = new CRC32 ( ) ; crc32 . update ( memory . array ( ) , headerLength , entryLength ) ; long checksum = crc32 . getValue ( ) ; // Rewind the in-memory buffer and write the length, checksum, and offset. memory . rewind ( ) . writeUnsignedInt ( checksum ) . writeLong ( offset ) ; // If the term has not yet been written, write the term to this entry. if ( skipTerm ) { memory . writeBoolean ( false ) ; } else { memory . writeBoolean ( true ) . writeLong ( term ) ; } // Write the entry length and entry to the segment. buffer . writeInt ( totalLength ) . write ( memory . rewind ( ) ) ; // Index the offset, position, and length. offsetIndex . index ( offset , position ) ; // If the entry term is greater than the last indexed term, index the term. if ( term > lastTerm ) { termIndex . index ( offset , term ) ; } // Reset skip to zero since we wrote a new entry. skip = 0 ; return index ; }
Commits an entry to the segment .