Update near cache to support synchronous listeners (#98)

* Update near cache to support syncrhonous listeners
* Add configurable pruneFactor
* Minor review changes
* Address test failure

Author: Tim Middleton

coherence/coherence_test_helpers.go

@@ -122,6 +122,15 @@ func GetSessionQueueID(session *Session, queue string) *int32 {
 	return session.getQueueID(queue)
 }
 
+func GetNearCachePruneFactor[K comparable, V any](namedMap NamedMap[K, V]) float32 {
+	ncOptions := namedMap.getBaseClient().cacheOpts.NearCacheOptions
+	if ncOptions == nil {
+		return 0.0
+	}
+
+	return ncOptions.PruneFactor
+}
+
 // revive:disable:unexported-return
 func GetKeyListenerGroupMap[K comparable, V any](namedMap NamedMap[K, V]) map[K]*listenerGroupV1[K, V] {
 	return namedMap.getBaseClient().keyListenersV1

coherence/common.go

@@ -86,7 +86,7 @@ type baseClient[K comparable, V any] struct {
 	mutex                      *sync.RWMutex
 	nearCache                  *localCacheImpl[K, V]
 	nearCacheListener          *namedCacheNearCacheListener[K, V]
-	nearCacheLifecycleListener *namedCacheNearLifecyleListener[K, V]
+	nearCacheLifecycleListener *namedCacheNearLifestyleListener[K, V]
 
 	// gRPC v1 listeners registered
 	keyListenersV1       map[K]*listenerGroupV1[K, V]
@@ -103,15 +103,26 @@ type CacheOptions struct {
 
 // NearCacheOptions defines options when creating a near cache.
 type NearCacheOptions struct {
-	TTL                  time.Duration
-	HighUnits            int64
-	HighUnitsMemory      int64
+	// TTL is the maximum time to keep the entry in the near cache. When this time has been reached it will be expired.
+	TTL time.Duration
+
+	// HighUnits is the maximum number of cache entries to keep in the near cache.
+	HighUnits int64
+
+	// HighUnitsMemory is the maximum amount of memory to use for entries in the near cache.
+	HighUnitsMemory int64
+
 	InvalidationStrategy InvalidationStrategyType // currently only supports ListenAll
+
+	// PruneFactor indicates the percentage of the total number of units that will remain
+	// after the cache manager prunes the near cache(i.e. this is the "low watermark" value)
+	// this value is in the range 0.1 to 1.0 and the default is 0.8 or 80%.
+	PruneFactor float32
 }
 
 func (n NearCacheOptions) String() string {
-	return fmt.Sprintf("NearCacheOptions{TTL=%v, HighUnits=%v, HighUnitsMemory=%v, invalidationStrategy=%v}",
-		n.TTL, n.HighUnits, n.HighUnitsMemory, getInvalidationStrategyString(n.InvalidationStrategy))
+	return fmt.Sprintf("NearCacheOptions{TTL=%v, HighUnits=%v, HighUnitsMemory=%v, PruneFactor=%.2f, invalidationStrategy=%v}",
+		n.TTL, n.HighUnits, n.HighUnitsMemory, n.PruneFactor, getInvalidationStrategyString(n.InvalidationStrategy))
 }
 
 // WithExpiry returns a function to set the default expiry for a [NamedCache]. This option is not valid on [NamedMap].
@@ -134,6 +145,7 @@ func executeClear[K comparable, V any](ctx context.Context, bc *baseClient[K, V]
 		err       = bc.ensureClientConnection()
 		nearCache = bc.nearCache
 	)
+
 	if err != nil {
 		return err
 	}
@@ -152,7 +164,7 @@ func executeClear[K comparable, V any](ctx context.Context, bc *baseClient[K, V]
 	}
 
 	// clear the near cache
-	if nearCache != nil {
+	if bc.session.GetProtocolVersion() == 0 && nearCache != nil {
 		nearCache.Clear()
 	}
 
@@ -272,7 +284,7 @@ func executeTruncate[K comparable, V any](ctx context.Context, bc *baseClient[K,
 		_, err = bc.client.Truncate(newCtx, &request)
 	}
 
-	// clear the near cache
+	// clear the near cache as the lifecycle listeners are not synchronous
 	if nearCache != nil {
 		nearCache.Clear()
 	}
@@ -1396,8 +1408,8 @@ func executePutAll[K comparable, V any](ctx context.Context, bc *baseClient[K, V
 		}
 	}
 
-	// if we have near cache and the entry exists then update
-	if nearCache != nil {
+	// if we have near cache and the entry exists then update (gRPC v0)
+	if bc.session.GetProtocolVersion() == 0 && nearCache != nil {
 		for k, v := range entries {
 			if oldVal := nearCache.Get(k); oldVal != nil {
 				nearCache.Put(k, v)
@@ -1510,8 +1522,8 @@ func executePutWithExpiry[K comparable, V any](ctx context.Context, bc *baseClie
 	}
 
 	// if we have near cache and the entry exists then update this as well because we do
-	// not use synchronous listener like we do on Java
-	if nearCache != nil {
+	// not use synchronous listener in gRPC v0
+	if bc.session.GetProtocolVersion() == 0 && nearCache != nil {
 		if oldValue := nearCache.Get(key); oldValue != nil {
 			nearCache.Put(key, value)
 		}
@@ -1580,7 +1592,7 @@ func executeRemove[K comparable, V any](ctx context.Context, bc *baseClient[K, V
 		}
 	}
 
-	if nearCache != nil {
+	if bc.session.GetProtocolVersion() == 0 && nearCache != nil {
 		nearCache.Remove(key)
 	}
 
@@ -1632,7 +1644,7 @@ func executeRemoveMapping[K comparable, V any](ctx context.Context, bc *baseClie
 		}
 	}
 
-	if result.Value && nearCache != nil {
+	if bc.session.GetProtocolVersion() == 0 && result.Value && nearCache != nil {
 		nearCache.Remove(key)
 	}
 
@@ -1738,7 +1750,7 @@ func executeReplaceMapping[K comparable, V any](ctx context.Context, bc *baseCli
 		}
 	}
 
-	if nearCache != nil && result.Value {
+	if bc.session.GetProtocolVersion() == 0 && nearCache != nil && result.Value {
 		if old := nearCache.Get(key); old != nil {
 			nearCache.Put(key, newValue)
 		}

coherence/doc.go

@@ -774,9 +774,13 @@ The following example shows how to get a named cache that will cache entries fro
 2. Creating a Near Cache specifying maximum number of entries to store
 
 The following example shows how to get a named cache that will cache up to 100 entries from Get() or GetAll().
-When the threshold of HighUnits is reached, the near cache is pruned to 80% of its size and evicts least recently
+When the threshold of HighUnits is reached, the near cache is pruned to the default of 80% of its size and evicts least recently
 accessed and created entries.
 
+Note: The default prune percentage is 0.8 (80%) which indicates the percentage of the total number of
+units that will remain after the cache manager prunes the near cache( i.e. this is the "low watermark" value).
+This can be changed by setting the PruneFactory to a value in the range 0.1 to 1.0 in [NearCacheOptions].
+
 	// specify HighUnits of 1000
 	nearCacheOptions := coherence.NearCacheOptions{HighUnits: 1000}
 
@@ -827,8 +831,8 @@ accessed and created entries.
 
 	// print the near cache stats via String()
 	fmt.Println(namedMap.GetNearCacheStats())
-	// localCache{name=customers options=localCacheOptions{ttl=0s, highUnits=0, highUnitsMemory=10.0KB, invalidation=ListenAll},
-	// stats=CacheStats{puts=5000, gets=5000, hits=0, misses=5000, missesDuration=4.95257111s, hitRate=0, prunes=7, prunesDuration=196.498µs, size=398, memoryUsed=9.3KB}}
+	// localCache{name=my-near-cache-high-units, options=localCacheOptions{ttl=0s, highUnits=1000, highUnitsMemory=0B, pruneFactor=0.80, invalidation=ListenAll}, stats=CacheStats{puts=1001, gets=1002, hits=1, misses=1001, missesDuration=4.628931138s,
+	// hitRate=0.0998004, prunes=1, prunesDuration=181.533µs, expires=0, expiresDuration=0s, size=200, memoryUsed=53.2KB}}
 
 [Coherence Documentation]: https://docs.oracle.com/en/middleware/standalone/coherence/14.1.1.2206/develop-applications/introduction-coherence-caches.html
 [examples]: https://github.com/oracle/coherence-go-client/tree/main/examples

coherence/event.go

@@ -471,21 +471,42 @@ func (l *mapLifecycleListener[K, V]) OnAny(callback func(MapLifecycleEvent[K, V]
 // MapListener allows registering callbacks to be notified when mutations events
 // occur within a [NamedMap] or [NamedCache].
 type MapListener[K comparable, V any] interface {
+	// OnInserted registers a callback that will be notified when an entry is inserted.
 	OnInserted(callback func(MapEvent[K, V])) MapListener[K, V]
+
+	// OnUpdated registers a callback that will be notified when an entry is updated.
 	OnUpdated(callback func(MapEvent[K, V])) MapListener[K, V]
+
+	// OnDeleted registers a callback that will be notified when an entry is deleted.
 	OnDeleted(callback func(MapEvent[K, V])) MapListener[K, V]
+
+	// OnAny registers a callback that will be notified when any entry mutation has occurred.
 	OnAny(callback func(MapEvent[K, V])) MapListener[K, V]
 	dispatch(event MapEvent[K, V])
+
+	// SetSynchronous sets this [MapListener] as synchronous.
+	SetSynchronous()
+
+	// SetPriming sets this [MapListener] as priming.
+	SetPriming()
+
+	// IsSynchronous indicates this [MapListener] is synchronous.
+	IsSynchronous() bool
+
+	// IsPriming indicates this [MapListener] is priming.
+	IsPriming() bool
 }
 
 // mapListener struct containing data members to satisfy the [MapListener] contract.
 type mapListener[K comparable, V any] struct {
-	emitter *eventEmitter[MapEventType, MapEvent[K, V]]
+	emitter     *eventEmitter[MapEventType, MapEvent[K, V]]
+	synchronous bool
+	priming     bool
 }
 
 // NewMapListener creates and returns a pointer to a new [MapListener] instance.
 func NewMapListener[K comparable, V any]() MapListener[K, V] {
-	return &mapListener[K, V]{newEventEmitter[MapEventType, MapEvent[K, V]]()}
+	return &mapListener[K, V]{emitter: newEventEmitter[MapEventType, MapEvent[K, V]]()}
 }
 
 // dispatch dispatches the specified event to the appropriate group of listeners.
@@ -499,6 +520,26 @@ func (l *mapListener[K, V]) on(event MapEventType, callback func(MapEvent[K, V])
 	return l
 }
 
+// SetSynchronous sets this listener to be synchronous.
+func (l *mapListener[K, V]) SetSynchronous() {
+	l.synchronous = true
+}
+
+// SetPriming sets this listener to be a priming listener.
+func (l *mapListener[K, V]) SetPriming() {
+	l.priming = true
+}
+
+// IsSynchronous indicates if this listener is synchronous.
+func (l *mapListener[K, V]) IsSynchronous() bool {
+	return l.synchronous
+}
+
+// IsPriming indicates if this listener is a priming listener.
+func (l *mapListener[K, V]) IsPriming() bool {
+	return l.priming
+}
+
 // OnInserted registers a callback that will be notified when an entry is inserted.
 func (l *mapListener[K, V]) OnInserted(callback func(MapEvent[K, V])) MapListener[K, V] {
 	return l.on(EntryInserted, callback)