-
Notifications
You must be signed in to change notification settings - Fork 213
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Core: state machine entity base class (#1511)
* Core: state machine entity base class * Update CHANGELOG.md * Update EntitySendRetryManagerTest.java * Update StateMachine.java * Renamed StateMachine to StatefulEntity * Update ContractNegotiationIntegrationTest.java * grammar fix
- Loading branch information
Showing
18 changed files
with
386 additions
and
384 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
91 changes: 91 additions & 0 deletions
91
...c/main/java/org/eclipse/dataspaceconnector/core/defaults/InMemoryStatefulEntityStore.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| /* | ||
| * Copyright (c) 2022 Microsoft Corporation | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Apache License, Version 2.0 which is available at | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * Contributors: | ||
| * Microsoft Corporation - initial API and implementation | ||
| * | ||
| */ | ||
|
|
||
| package org.eclipse.dataspaceconnector.core.defaults; | ||
|
|
||
| import org.eclipse.dataspaceconnector.common.concurrency.LockManager; | ||
| import org.eclipse.dataspaceconnector.spi.entity.StatefulEntity; | ||
| import org.eclipse.dataspaceconnector.spi.query.QueryResolver; | ||
| import org.eclipse.dataspaceconnector.spi.query.QuerySpec; | ||
| import org.eclipse.dataspaceconnector.spi.query.ReflectionBasedQueryResolver; | ||
| import org.jetbrains.annotations.NotNull; | ||
|
|
||
| import java.util.Comparator; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.locks.ReentrantReadWriteLock; | ||
| import java.util.stream.Stream; | ||
|
|
||
| import static java.util.stream.Collectors.toList; | ||
|
|
||
| /** | ||
| * An in-memory, threadsafe entity store for a {@link StatefulEntity}. | ||
| * This implementation is intended for testing purposes only. | ||
| */ | ||
| public class InMemoryStatefulEntityStore<T extends StatefulEntity<T>> { | ||
| private final Map<String, Item<T>> entitiesById = new ConcurrentHashMap<>(); | ||
| private final QueryResolver<T> queryResolver; | ||
| private final LockManager lockManager = new LockManager(new ReentrantReadWriteLock()); | ||
|
|
||
| public InMemoryStatefulEntityStore(Class<T> clazz) { | ||
| queryResolver = new ReflectionBasedQueryResolver<>(clazz); | ||
| } | ||
|
|
||
| public T find(String id) { | ||
| var t = entitiesById.get(id); | ||
| if (t == null) { | ||
| return null; | ||
| } | ||
| return t.item.copy(); | ||
| } | ||
|
|
||
| public void upsert(T entity) { | ||
| entitiesById.put(entity.getId(), new Item<>(entity.copy())); | ||
| } | ||
|
|
||
| public void delete(String id) { | ||
| entitiesById.remove(id); | ||
| } | ||
|
|
||
| public Stream<T> findAll(QuerySpec querySpec) { | ||
| return queryResolver.query(findAll(), querySpec); | ||
| } | ||
|
|
||
| public @NotNull List<T> nextForState(int state, int max) { | ||
| return lockManager.writeLock(() -> { | ||
| var items = entitiesById.values().stream() | ||
| .filter(e -> e.item.getState() == state) | ||
| .filter(e -> !e.leased) | ||
| .sorted(Comparator.comparingLong(e -> e.item.getStateTimestamp())) //order by state timestamp, oldest first | ||
| .limit(max) | ||
| .collect(toList()); | ||
| items.forEach(e -> e.leased = true); | ||
| return items.stream().map(e -> e.item.copy()).collect(toList()); | ||
| }); | ||
| } | ||
|
|
||
| public Stream<T> findAll() { | ||
| return entitiesById.values().stream().map(e -> e.item); | ||
| } | ||
|
|
||
| private static class Item<V> { | ||
| private final V item; | ||
| private boolean leased = false; | ||
|
|
||
| Item(V item) { | ||
| this.item = item; | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.