mirror of https://github.com/hyperledger/besu
NC-1721: Filter timeout if not queried for 10 minutes (#66)
Lucas Saldanha
6 years ago
committed by
GitHub
parent
793f149f96
commit
03f92e3eb0
@ -0,0 +1,28 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.core.Hash; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
|
||||
/** Tracks new blocks being added to the blockchain. */ |
||||
class BlockFilter extends Filter { |
||||
|
||||
private final List<Hash> blockHashes = new ArrayList<>(); |
||||
|
||||
BlockFilter(final String id) { |
||||
super(id); |
||||
} |
||||
|
||||
void addBlockHash(final Hash hash) { |
||||
blockHashes.add(hash); |
||||
} |
||||
|
||||
List<Hash> blockHashes() { |
||||
return blockHashes; |
||||
} |
||||
|
||||
void clearBlockHashes() { |
||||
blockHashes.clear(); |
||||
} |
||||
} |
||||
@ -0,0 +1,41 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import java.time.Duration; |
||||
import java.time.Instant; |
||||
|
||||
import com.google.common.annotations.VisibleForTesting; |
||||
|
||||
abstract class Filter { |
||||
|
||||
private static final Duration DEFAULT_EXPIRE_DURATION = Duration.ofMinutes(10); |
||||
|
||||
private final String id; |
||||
private Instant expireTime; |
||||
|
||||
Filter(final String id) { |
||||
this.id = id; |
||||
resetExpireTime(); |
||||
} |
||||
|
||||
String getId() { |
||||
return id; |
||||
} |
||||
|
||||
void resetExpireTime() { |
||||
this.expireTime = Instant.now().plus(DEFAULT_EXPIRE_DURATION); |
||||
} |
||||
|
||||
boolean isExpired() { |
||||
return Instant.now().isAfter(expireTime); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
void setExpireTime(final Instant expireTime) { |
||||
this.expireTime = expireTime; |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
Instant getExpireTime() { |
||||
return expireTime; |
||||
} |
||||
} |
||||
@ -0,0 +1,72 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.Collection; |
||||
import java.util.Map; |
||||
import java.util.Optional; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.stream.Collectors; |
||||
import java.util.stream.Stream; |
||||
|
||||
public class FilterRepository { |
||||
|
||||
private final Map<String, Filter> filters = new ConcurrentHashMap<>(); |
||||
|
||||
public FilterRepository() {} |
||||
|
||||
Collection<Filter> getFilters() { |
||||
return new ArrayList<>(filters.values()); |
||||
} |
||||
|
||||
<T extends Filter> Collection<T> getFiltersOfType(final Class<T> filterClass) { |
||||
return filters |
||||
.values() |
||||
.stream() |
||||
.flatMap(f -> getIfTypeMatches(f, filterClass).map(Stream::of).orElseGet(Stream::empty)) |
||||
.collect(Collectors.toList()); |
||||
} |
||||
|
||||
<T extends Filter> Optional<T> getFilter(final String filterId, final Class<T> filterClass) { |
||||
final Filter filter = filters.get(filterId); |
||||
return getIfTypeMatches(filter, filterClass); |
||||
} |
||||
|
||||
@SuppressWarnings("unchecked") |
||||
private <T extends Filter> Optional<T> getIfTypeMatches( |
||||
final Filter filter, final Class<T> filterClass) { |
||||
if (filter == null) { |
||||
return Optional.empty(); |
||||
} |
||||
|
||||
if (!filterClass.isAssignableFrom(filter.getClass())) { |
||||
return Optional.empty(); |
||||
} |
||||
|
||||
return Optional.of((T) filter); |
||||
} |
||||
|
||||
boolean exists(final String id) { |
||||
return filters.containsKey(id); |
||||
} |
||||
|
||||
void save(final Filter filter) { |
||||
if (filter == null) { |
||||
throw new IllegalArgumentException("Can't save null filter"); |
||||
} |
||||
|
||||
if (exists(filter.getId())) { |
||||
throw new IllegalArgumentException( |
||||
String.format("Filter with id %s already exists", filter.getId())); |
||||
} |
||||
|
||||
filters.put(filter.getId(), filter); |
||||
} |
||||
|
||||
void delete(final String id) { |
||||
filters.remove(id); |
||||
} |
||||
|
||||
void deleteAll() { |
||||
filters.clear(); |
||||
} |
||||
} |
||||
@ -0,0 +1,21 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
class FilterTimeoutMonitor { |
||||
|
||||
private final FilterRepository filterRepository; |
||||
|
||||
FilterTimeoutMonitor(final FilterRepository filterRepository) { |
||||
this.filterRepository = filterRepository; |
||||
} |
||||
|
||||
void checkFilters() { |
||||
filterRepository |
||||
.getFilters() |
||||
.forEach( |
||||
filter -> { |
||||
if (filter.isExpired()) { |
||||
filterRepository.delete(filter.getId()); |
||||
} |
||||
}); |
||||
} |
||||
} |
||||
@ -0,0 +1,51 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.internal.parameters.BlockParameter; |
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.internal.queries.LogWithMetadata; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
|
||||
class LogFilter extends Filter { |
||||
|
||||
private final BlockParameter fromBlock; |
||||
private final BlockParameter toBlock; |
||||
private final LogsQuery logsQuery; |
||||
|
||||
private final List<LogWithMetadata> logs = new ArrayList<>(); |
||||
|
||||
LogFilter( |
||||
final String id, |
||||
final BlockParameter fromBlock, |
||||
final BlockParameter toBlock, |
||||
final LogsQuery logsQuery) { |
||||
super(id); |
||||
this.fromBlock = fromBlock; |
||||
this.toBlock = toBlock; |
||||
this.logsQuery = logsQuery; |
||||
} |
||||
|
||||
public BlockParameter getFromBlock() { |
||||
return fromBlock; |
||||
} |
||||
|
||||
public BlockParameter getToBlock() { |
||||
return toBlock; |
||||
} |
||||
|
||||
public LogsQuery getLogsQuery() { |
||||
return logsQuery; |
||||
} |
||||
|
||||
void addLog(final List<LogWithMetadata> logs) { |
||||
this.logs.addAll(logs); |
||||
} |
||||
|
||||
List<LogWithMetadata> logs() { |
||||
return logs; |
||||
} |
||||
|
||||
void clearLogs() { |
||||
logs.clear(); |
||||
} |
||||
} |
||||
@ -0,0 +1,28 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.core.Hash; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
|
||||
/** Tracks new pending transactions that have arrived in the transaction pool */ |
||||
class PendingTransactionFilter extends Filter { |
||||
|
||||
private final List<Hash> transactionHashes = new ArrayList<>(); |
||||
|
||||
PendingTransactionFilter(final String id) { |
||||
super(id); |
||||
} |
||||
|
||||
void addTransactionHash(final Hash hash) { |
||||
transactionHashes.add(hash); |
||||
} |
||||
|
||||
List<Hash> transactionHashes() { |
||||
return transactionHashes; |
||||
} |
||||
|
||||
void clearTransactionHashes() { |
||||
transactionHashes.clear(); |
||||
} |
||||
} |
||||
@ -0,0 +1,191 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.assertj.core.api.Assertions.catchThrowable; |
||||
|
||||
import java.util.Collection; |
||||
import java.util.Optional; |
||||
|
||||
import org.assertj.core.util.Lists; |
||||
import org.junit.Before; |
||||
import org.junit.Test; |
||||
|
||||
public class FilterRepositoryTest { |
||||
|
||||
private FilterRepository repository; |
||||
|
||||
@Before |
||||
public void before() { |
||||
repository = new FilterRepository(); |
||||
} |
||||
|
||||
@Test |
||||
public void getFiltersShouldReturnAllFilters() { |
||||
BlockFilter filter1 = new BlockFilter("foo"); |
||||
BlockFilter filter2 = new BlockFilter("bar"); |
||||
repository.save(filter1); |
||||
repository.save(filter2); |
||||
|
||||
Collection<Filter> filters = repository.getFilters(); |
||||
|
||||
assertThat(filters).containsExactlyInAnyOrderElementsOf(Lists.newArrayList(filter1, filter2)); |
||||
} |
||||
|
||||
@Test |
||||
public void getFiltersShouldReturnEmptyListWhenRepositoryIsEmpty() { |
||||
assertThat(repository.getFilters()).isEmpty(); |
||||
} |
||||
|
||||
@Test |
||||
public void saveShouldAddFilterToRepository() { |
||||
BlockFilter filter = new BlockFilter("id"); |
||||
repository.save(filter); |
||||
|
||||
BlockFilter retrievedFilter = repository.getFilter("id", BlockFilter.class).get(); |
||||
|
||||
assertThat(retrievedFilter).isEqualToComparingFieldByField(filter); |
||||
} |
||||
|
||||
@Test |
||||
public void saveNullFilterShouldFail() { |
||||
Throwable throwable = catchThrowable(() -> repository.save(null)); |
||||
|
||||
assertThat(throwable) |
||||
.isInstanceOf(IllegalArgumentException.class) |
||||
.hasMessage("Can't save null filter"); |
||||
} |
||||
|
||||
@Test |
||||
public void saveFilterWithSameIdShouldFail() { |
||||
BlockFilter filter = new BlockFilter("x"); |
||||
repository.save(filter); |
||||
|
||||
Throwable throwable = catchThrowable(() -> repository.save(filter)); |
||||
|
||||
assertThat(throwable) |
||||
.isInstanceOf(IllegalArgumentException.class) |
||||
.hasMessage("Filter with id x already exists"); |
||||
} |
||||
|
||||
@Test |
||||
public void getSingleFilterShouldReturnExistingFilterOfCorrectType() { |
||||
BlockFilter filter = new BlockFilter("id"); |
||||
repository.save(filter); |
||||
|
||||
Optional<BlockFilter> optional = repository.getFilter(filter.getId(), BlockFilter.class); |
||||
|
||||
assertThat(optional.isPresent()).isTrue(); |
||||
assertThat(optional.get()).isEqualToComparingFieldByField(filter); |
||||
} |
||||
|
||||
@Test |
||||
public void getSingleFilterShouldReturnEmptyForFilterOfIncorrectType() { |
||||
BlockFilter filter = new BlockFilter("id"); |
||||
repository.save(filter); |
||||
|
||||
Optional<PendingTransactionFilter> optional = |
||||
repository.getFilter(filter.getId(), PendingTransactionFilter.class); |
||||
|
||||
assertThat(optional.isPresent()).isFalse(); |
||||
} |
||||
|
||||
@Test |
||||
public void getSingleFilterShouldReturnEmptyForAbsentId() { |
||||
BlockFilter filter = new BlockFilter("foo"); |
||||
repository.save(filter); |
||||
|
||||
Optional<BlockFilter> optional = repository.getFilter("bar", BlockFilter.class); |
||||
|
||||
assertThat(optional.isPresent()).isFalse(); |
||||
} |
||||
|
||||
@Test |
||||
public void getSingleFilterShouldReturnEmptyForEmptyRepository() { |
||||
Optional<BlockFilter> optional = repository.getFilter("id", BlockFilter.class); |
||||
|
||||
assertThat(optional.isPresent()).isFalse(); |
||||
} |
||||
|
||||
@Test |
||||
public void getFilterCollectionShouldReturnAllFiltersOfSpecificType() { |
||||
BlockFilter blockFilter1 = new BlockFilter("foo"); |
||||
BlockFilter blockFilter2 = new BlockFilter("biz"); |
||||
PendingTransactionFilter pendingTxFilter1 = new PendingTransactionFilter("bar"); |
||||
|
||||
Collection<BlockFilter> expectedFilters = Lists.newArrayList(blockFilter1, blockFilter2); |
||||
|
||||
repository.save(blockFilter1); |
||||
repository.save(blockFilter2); |
||||
repository.save(pendingTxFilter1); |
||||
|
||||
Collection<BlockFilter> blockFilters = repository.getFiltersOfType(BlockFilter.class); |
||||
|
||||
assertThat(blockFilters).containsExactlyInAnyOrderElementsOf(expectedFilters); |
||||
} |
||||
|
||||
@Test |
||||
public void getFilterCollectionShouldReturnEmptyForNoneMatchingTypes() { |
||||
PendingTransactionFilter filter = new PendingTransactionFilter("foo"); |
||||
repository.save(filter); |
||||
|
||||
Collection<BlockFilter> filters = repository.getFiltersOfType(BlockFilter.class); |
||||
|
||||
assertThat(filters).isEmpty(); |
||||
} |
||||
|
||||
@Test |
||||
public void getFilterCollectionShouldReturnEmptyListForEmptyRepository() { |
||||
Collection<BlockFilter> filters = repository.getFiltersOfType(BlockFilter.class); |
||||
|
||||
assertThat(filters).isEmpty(); |
||||
} |
||||
|
||||
@Test |
||||
public void existsShouldReturnTrueForExistingId() { |
||||
BlockFilter filter = new BlockFilter("id"); |
||||
repository.save(filter); |
||||
|
||||
assertThat(repository.exists("id")).isTrue(); |
||||
} |
||||
|
||||
@Test |
||||
public void existsShouldReturnFalseForAbsentId() { |
||||
BlockFilter filter = new BlockFilter("foo"); |
||||
repository.save(filter); |
||||
|
||||
assertThat(repository.exists("bar")).isFalse(); |
||||
} |
||||
|
||||
@Test |
||||
public void existsShouldReturnFalseForEmptyRepository() { |
||||
assertThat(repository.exists("id")).isFalse(); |
||||
} |
||||
|
||||
@Test |
||||
public void deleteExistingFilterShouldDeleteSuccessfully() { |
||||
BlockFilter filter = new BlockFilter("foo"); |
||||
repository.save(filter); |
||||
repository.delete(filter.getId()); |
||||
|
||||
assertThat(repository.exists(filter.getId())).isFalse(); |
||||
} |
||||
|
||||
@Test |
||||
public void deleteAbsentFilterDoesNothing() { |
||||
assertThat(repository.exists("foo")).isFalse(); |
||||
repository.delete("foo"); |
||||
} |
||||
|
||||
@Test |
||||
public void deleteAllShouldClearFilters() { |
||||
BlockFilter filter1 = new BlockFilter("foo"); |
||||
BlockFilter filter2 = new BlockFilter("biz"); |
||||
repository.save(filter1); |
||||
repository.save(filter2); |
||||
|
||||
repository.deleteAll(); |
||||
|
||||
assertThat(repository.exists(filter1.getId())).isFalse(); |
||||
assertThat(repository.exists(filter2.getId())).isFalse(); |
||||
} |
||||
} |
||||
@ -0,0 +1,36 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
||||
import java.time.Duration; |
||||
import java.time.Instant; |
||||
|
||||
import org.junit.Test; |
||||
|
||||
public class FilterTest { |
||||
|
||||
@Test |
||||
public void filterJustCreatedShouldNotBeExpired() { |
||||
BlockFilter filter = new BlockFilter("foo"); |
||||
|
||||
assertThat(filter.isExpired()).isFalse(); |
||||
} |
||||
|
||||
@Test |
||||
public void isExpiredShouldReturnTrueForExpiredFilter() { |
||||
BlockFilter filter = new BlockFilter("foo"); |
||||
filter.setExpireTime(Instant.now().minusSeconds(1)); |
||||
|
||||
assertThat(filter.isExpired()).isTrue(); |
||||
} |
||||
|
||||
@Test |
||||
public void resetExpireDateShouldIncrementExpireDate() { |
||||
BlockFilter filter = new BlockFilter("foo"); |
||||
filter.setExpireTime(Instant.now().minus(Duration.ofDays(1))); |
||||
filter.resetExpireTime(); |
||||
|
||||
assertThat(filter.getExpireTime()) |
||||
.isBeforeOrEqualTo(Instant.now().plus(Duration.ofMinutes(10))); |
||||
} |
||||
} |
||||
@ -0,0 +1,64 @@ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.internal.filter; |
||||
|
||||
import static org.mockito.Mockito.mock; |
||||
import static org.mockito.Mockito.spy; |
||||
import static org.mockito.Mockito.verify; |
||||
import static org.mockito.Mockito.verifyNoMoreInteractions; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
import java.util.Collections; |
||||
|
||||
import com.google.common.collect.Lists; |
||||
import org.junit.Before; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.mockito.Mock; |
||||
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
||||
@RunWith(MockitoJUnitRunner.class) |
||||
public class FilterTimeoutMonitorTest { |
||||
|
||||
@Mock private FilterRepository filterRepository; |
||||
|
||||
private FilterTimeoutMonitor timeoutMonitor; |
||||
|
||||
@Before |
||||
public void before() { |
||||
timeoutMonitor = new FilterTimeoutMonitor(filterRepository); |
||||
} |
||||
|
||||
@Test |
||||
public void expiredFilterShouldBeDeleted() { |
||||
Filter filter = spy(new BlockFilter("foo")); |
||||
when(filter.isExpired()).thenReturn(true); |
||||
when(filterRepository.getFilters()).thenReturn(Lists.newArrayList(filter)); |
||||
|
||||
timeoutMonitor.checkFilters(); |
||||
|
||||
verify(filterRepository).getFilters(); |
||||
verify(filterRepository).delete("foo"); |
||||
verifyNoMoreInteractions(filterRepository); |
||||
} |
||||
|
||||
@Test |
||||
public void nonExpiredFilterShouldNotBeDeleted() { |
||||
Filter filter = mock(Filter.class); |
||||
when(filter.isExpired()).thenReturn(false); |
||||
when(filterRepository.getFilters()).thenReturn(Lists.newArrayList(filter)); |
||||
|
||||
timeoutMonitor.checkFilters(); |
||||
|
||||
verify(filter).isExpired(); |
||||
verifyNoMoreInteractions(filter); |
||||
} |
||||
|
||||
@Test |
||||
public void checkEmptyFilterRepositoryDoesNothing() { |
||||
when(filterRepository.getFilters()).thenReturn(Collections.emptyList()); |
||||
|
||||
timeoutMonitor.checkFilters(); |
||||
|
||||
verify(filterRepository).getFilters(); |
||||
verifyNoMoreInteractions(filterRepository); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue