private CommitResult commitTransaction ( TransactionScope < K > transactionScope ) { TransactionLog < MapUpdate < K , byte [ ] > > transactionLog = transactionScope . transactionLog ( ) ; boolean retainTombstones = ! activeTransactions . isEmpty ( ) ; List < AtomicMapEvent < K , byte [ ] > > eventsToPublish = Lists . newArrayList ( ) ; for ( MapUpdate < K , byte [ ] > record : transactionLog . records ( ) ) { if ( record . type ( ) == MapUpdate . Type . VERSION_MATCH ) { continue ; } K key = record . key ( ) ; checkState ( preparedKeys . remove ( key ) , "key is not prepared" ) ; if ( record . type ( ) == MapUpdate . Type . LOCK ) { continue ; } MapEntryValue previousValue = entries ( ) . remove ( key ) ; // Cancel the previous timer if set. cancelTtl ( previousValue ) ; MapEntryValue newValue = null ; // If the record is not a delete, create a transactional commit. if ( record . type ( ) != MapUpdate . Type . REMOVE_IF_VERSION_MATCH ) { newValue = new MapEntryValue ( MapEntryValue . Type . VALUE , currentVersion , record . value ( ) , 0 , 0 ) ; } else if ( retainTombstones ) { // For deletes, if tombstones need to be retained then create and store a tombstone commit. newValue = new MapEntryValue ( MapEntryValue . Type . TOMBSTONE , currentVersion , null , 0 , 0 ) ; } AtomicMapEvent < K , byte [ ] > event ; if ( newValue != null ) { entries ( ) . put ( key , newValue ) ; if ( ! valueIsNull ( newValue ) ) { if ( ! valueIsNull ( previousValue ) ) { event = new AtomicMapEvent <> ( AtomicMapEvent . Type . UPDATE , key , toVersioned ( newValue ) , toVersioned ( previousValue ) ) ; } else { event = new AtomicMapEvent <> ( AtomicMapEvent . Type . INSERT , key , toVersioned ( newValue ) , null ) ; } } else { event = new AtomicMapEvent <> ( AtomicMapEvent . Type . REMOVE , key , null , toVersioned ( previousValue ) ) ; } } else { event = new AtomicMapEvent <> ( AtomicMapEvent . Type . REMOVE , key , null , toVersioned ( previousValue ) ) ; } eventsToPublish . add ( event ) ; } publish ( eventsToPublish ) ; return CommitResult . OK ; }
Applies committed operations to the state machine .