import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Nonnull
@SuppressWarnings("unchecked")
public static <T, K, V> ProcessorMetaSupplier updateMapP(
        @Nonnull String name,
        @Nullable ClientConfig clientConfig,
        @Nonnull DistributedFunction<T, K> toKeyFn,
        @Nonnull DistributedFunction<T, EntryProcessor<K, V>> toEntryProcessorFn
) {
    boolean isLocal = clientConfig == null;
    return dontParallelize(new EntryProcessorWriterSupplier<>(
                    name,
                    serializableConfig(clientConfig),
                    toKeyFn,
                    toEntryProcessorFn,
                    isLocal
            )
    );
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Override
protected boolean tryProcess(int ordinal, @Nonnull Object object) throws Exception {
    checkError();
    if (!tryIncrement(numConcurrentOps, 1, MAX_PARALLEL_ASYNC_OPS)) {
        return false;
    }
    try {
        T item = (T) object;
        EntryProcessor<K, V> entryProcessor = toEntryProcessorFn.apply(item);
        K key = toKeyFn.apply(item);
        map.submitToKey(key, entryProcessor, callback);
        return true;
    } catch (HazelcastInstanceNotActiveException e) {
        handleInstanceNotActive(instance, e, isLocal);
        return false;
    }
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Test
public void saveUpdatedAttributeFlushModeImmediate() {
	verify(this.sessions, times(1)).addEntryListener(any(MapListener.class),
			anyBoolean());

	this.repository.setHazelcastFlushMode(HazelcastFlushMode.IMMEDIATE);

	HazelcastSession session = this.repository.createSession();
	session.setAttribute("testName", "testValue");
	verify(this.sessions, times(1)).set(eq(session.getId()),
			eq(session.getDelegate()), isA(Long.class), eq(TimeUnit.SECONDS));
	verify(this.sessions, times(1)).executeOnKey(eq(session.getId()),
			any(EntryProcessor.class));

	this.repository.save(session);
	verifyZeroInteractions(this.sessions);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Test
public void removeAttributeFlushModeImmediate() {
	verify(this.sessions, times(1)).addEntryListener(any(MapListener.class),
			anyBoolean());

	this.repository.setHazelcastFlushMode(HazelcastFlushMode.IMMEDIATE);

	HazelcastSession session = this.repository.createSession();
	session.removeAttribute("testName");
	verify(this.sessions, times(1)).set(eq(session.getId()),
			eq(session.getDelegate()), isA(Long.class), eq(TimeUnit.SECONDS));
	verify(this.sessions, times(1)).executeOnKey(eq(session.getId()),
			any(EntryProcessor.class));

	this.repository.save(session);
	verifyZeroInteractions(this.sessions);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Test
public void saveUpdatedLastAccessedTimeFlushModeImmediate() {
	verify(this.sessions, times(1)).addEntryListener(any(MapListener.class),
			anyBoolean());

	this.repository.setHazelcastFlushMode(HazelcastFlushMode.IMMEDIATE);

	HazelcastSession session = this.repository.createSession();
	session.setLastAccessedTime(Instant.now());
	verify(this.sessions, times(1)).set(eq(session.getId()),
			eq(session.getDelegate()), isA(Long.class), eq(TimeUnit.SECONDS));
	verify(this.sessions, times(1)).executeOnKey(eq(session.getId()),
			any(EntryProcessor.class));

	this.repository.save(session);
	verifyZeroInteractions(this.sessions);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Test
public void saveUpdatedMaxInactiveIntervalInSecondsFlushModeImmediate() {
	verify(this.sessions, times(1)).addEntryListener(any(MapListener.class),
			anyBoolean());

	this.repository.setHazelcastFlushMode(HazelcastFlushMode.IMMEDIATE);

	HazelcastSession session = this.repository.createSession();
	session.setMaxInactiveInterval(Duration.ofSeconds(1));
	verify(this.sessions, times(1)).set(eq(session.getId()),
			eq(session.getDelegate()), isA(Long.class), eq(TimeUnit.SECONDS));
	verify(this.sessions, times(1)).executeOnKey(eq(session.getId()),
			any(EntryProcessor.class));

	this.repository.save(session);
	verifyZeroInteractions(this.sessions);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * Execute on key.
 *
 * @param sessionId the session id
 * @param processor the processor
 * @return the object
 * @throws Exception
 */
Object executeOnKey(String sessionId, EntryProcessor processor) throws Exception {
    try {
        return clusterMap.executeOnKey(sessionId, processor);
    } catch (Exception e) {
        LOGGER.finest("Cannot connect hazelcast server", e);
        throw e;
    }
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * @deprecated not implemented yet
 * @throws UnsupportedOperationException not implemented yet
 */
@Deprecated
@SuppressWarnings("rawtypes")
@Override
public Object executeOnKey(K key, EntryProcessor entryProcessor) {
    throw new UnsupportedOperationException();
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * @deprecated not implemented yet
 * @throws UnsupportedOperationException not implemented yet
 */
@Deprecated
@SuppressWarnings("rawtypes")
@Override
public Map<K, Object> executeOnKeys(Set<K> keys,
        EntryProcessor entryProcessor) {
    throw new UnsupportedOperationException();
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * @deprecated not implemented yet
 * @throws UnsupportedOperationException not implemented yet
 */
@Deprecated
@SuppressWarnings("rawtypes")
@Override
public void submitToKey(K key, EntryProcessor entryProcessor,
        ExecutionCallback callback) {
    throw new UnsupportedOperationException();
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * @deprecated not implemented yet
 * @throws UnsupportedOperationException not implemented yet
 */
@Deprecated
@SuppressWarnings("rawtypes")
@Override
public Map<K, Object> executeOnEntries(EntryProcessor entryProcessor) {
    throw new UnsupportedOperationException();
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * Returns a sink equivalent to {@link #mapWithEntryProcessor}, but for a map
 * in a remote Hazelcast cluster identified by the supplied {@code
 * ClientConfig}.
 */
public static <E, K, V> Sink<E> remoteMapWithEntryProcessor(
        @Nonnull String mapName,
        @Nonnull ClientConfig clientConfig,
        @Nonnull DistributedFunction<E, K> toKeyFn,
        @Nonnull DistributedFunction<E, EntryProcessor<K, V>> toEntryProcessorFn

) {
    return fromProcessor("remoteMapWithEntryProcessorSink(" + mapName + ')',
            updateRemoteMapP(mapName, clientConfig, toKeyFn, toEntryProcessorFn));
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
private EntryProcessorWriter(HazelcastInstance instance, String name,
                             DistributedFunction toKeyFn,
                             DistributedFunction<T, EntryProcessor<K, V>> toEntryProcessorFn,
                             boolean isLocal) {
    this.instance = instance;
    this.map = instance.getMap(name);
    this.toKeyFn = toKeyFn;
    this.toEntryProcessorFn = toEntryProcessorFn;
    this.isLocal = isLocal;
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
private EntryProcessorWriterSupplier(String name, SerializableClientConfig clientConfig,
                                     DistributedFunction<T, K> toKeyFn,
                                     DistributedFunction<T, EntryProcessor<K, V>> toEntryProcessorFn,
                                     boolean isLocal) {
    this.name = name;
    this.clientConfig = clientConfig;
    this.toKeyFn = toKeyFn;
    this.toEntryProcessorFn = toEntryProcessorFn;
    this.isLocal = isLocal;
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
public static <K, V> EntryProcessor<K, V> entryProcessor(
        DistributedBiFunction<? super K, ? super V, ? extends V> remappingFunction
) {
    return new AbstractEntryProcessor<K, V>() {
        @Override
        public Object process(Entry<K, V> entry) {
            V newValue = remappingFunction.apply(entry.getKey(), entry.getValue());
            entry.setValue(newValue);
            return newValue;
        }
    };
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * Returns a supplier of processors for
 * {@link com.hazelcast.jet.Sinks#mapWithEntryProcessor(String, DistributedFunction, DistributedFunction)}.
 */
@Nonnull
public static <T, K, V> ProcessorMetaSupplier updateMapP(
        @Nonnull String mapName,
        @Nonnull DistributedFunction<T, K> toKeyFn,
        @Nonnull DistributedFunction<T, EntryProcessor<K, V>> toEntryProcessorFn

) {
    return HazelcastWriters.updateMapP(mapName, null, toKeyFn, toEntryProcessorFn);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
/**
 * Returns a supplier of processors for
 * {@link com.hazelcast.jet.Sinks#remoteMapWithEntryProcessor(String, ClientConfig, DistributedFunction,
 * DistributedFunction)}.
 */
@Nonnull
public static <T, K, V> ProcessorMetaSupplier updateRemoteMapP(
        @Nonnull String mapName,
        @Nonnull ClientConfig clientConfig,
        @Nonnull DistributedFunction<T, K> toKeyFn,
        @Nonnull DistributedFunction<T, EntryProcessor<K, V>> toEntryProcessorFn
) {
    return HazelcastWriters.updateMapP(mapName, clientConfig, toKeyFn, toEntryProcessorFn);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Override
public Object executeOnKey(K key, EntryProcessor entryProcessor) {
    return map.executeOnKey(key, entryProcessor);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Override
public Map<K, Object> executeOnKeys(Set<K> keys, EntryProcessor entryProcessor) {
    return map.executeOnKeys(keys, entryProcessor);
}
 

import com.hazelcast.map.EntryProcessor; //导入依赖的package包/类
@Override
public void submitToKey(K key, EntryProcessor entryProcessor, ExecutionCallback callback) {
    map.submitToKey(key, entryProcessor, callback);
}