package com.battery.business.service.impl;
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertFalse;
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
|
import java.time.Instant;
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import org.junit.jupiter.api.Test;
|
import com.battery.device.api.DeviceAdapter;
|
import com.battery.device.api.TestDataObservationAdapter;
|
import com.battery.device.api.model.AdapterCapabilities;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.model.DetectedChannel;
|
import com.battery.device.api.model.RealtimeSnapshot;
|
import com.battery.device.api.registry.AdapterInstanceConfig;
|
import com.battery.device.api.registry.AdapterRegistryException;
|
import com.battery.device.api.registry.DeviceAdapterRegistry;
|
|
/**
|
* 多来源实例：每个通道的取数请求只送到它所属的实例；实例缺失时明确失败，不回退到其他实例。
|
*/
|
class RoutingTestDataObservationAdapterTest
|
{
|
@Test
|
void sendsEachChannelRequestToTheInstanceNamedByItsAdapterCode()
|
{
|
DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
|
try
|
{
|
RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
|
RecordingAdapter first = new RecordingAdapter("neware76", 1L);
|
RecordingAdapter second = new RecordingAdapter("neware80", 2L);
|
registry.register(config("neware76"), first);
|
registry.register(config("neware80"), second);
|
|
assertEquals(1L, router.latestCabinet(channel("neware76")).sequenceId());
|
assertEquals(2L, router.latestCabinet(channel("neware80")).sequenceId());
|
assertEquals(List.of("neware76"), first.calls());
|
assertEquals(List.of("neware80"), second.calls());
|
assertEquals(60000L, router.persistenceIntervalMillis(channel("neware76")));
|
assertEquals(70000L, router.persistenceIntervalMillis(channel("neware80")));
|
}
|
finally
|
{
|
registry.close();
|
}
|
}
|
|
@Test
|
void missingInstanceFailsExplicitlyInsteadOfFallingBackToAnotherInstance()
|
{
|
DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
|
try
|
{
|
RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
|
RecordingAdapter only = new RecordingAdapter("neware76", 1L);
|
registry.register(config("neware76"), only);
|
|
// 实例缺失：明确失败，且没有落到已注册的实例上。
|
assertThrows(AdapterRegistryException.class, () -> router.latestCabinet(channel("neware80")));
|
assertThrows(AdapterRegistryException.class, () -> router.samplesSince(channel("neware80"), null));
|
assertThrows(AdapterRegistryException.class, () -> router.currentWatermark(channel("neware80")));
|
assertTrue(only.calls().isEmpty(), "缺失实例的请求不得落到其他实例");
|
}
|
finally
|
{
|
registry.close();
|
}
|
}
|
|
@Test
|
void aggregatedAdapterDoesNotPretendToDiscoverDevices()
|
{
|
DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
|
try
|
{
|
RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
|
assertThrows(UnsupportedOperationException.class, router::discoverChannels);
|
assertEquals(RoutingTestDataObservationAdapter.ADAPTER_CODE, router.adapterCode());
|
assertFalse(router.capabilities().control());
|
}
|
finally
|
{
|
registry.close();
|
}
|
}
|
|
private static ChannelIdentity channel(String adapterCode)
|
{
|
return new ChannelIdentity(adapterCode, 21, 36, 1, 2);
|
}
|
|
private static AdapterInstanceConfig config(String adapterCode)
|
{
|
return new AdapterInstanceConfig(adapterCode, "NEWARE_MYSQL", "127.0.0.1", 3306, 60000,
|
"Asia/Hong_Kong", true, AdapterCapabilities.observeOnly());
|
}
|
|
/** 记录被调用次数的实例替身；返回的序号用来区分请求落到了哪一个实例。 */
|
private static final class RecordingAdapter implements DeviceAdapter, TestDataObservationAdapter
|
{
|
private final String code;
|
private final long sequenceId;
|
private final List<String> calls = new ArrayList<>();
|
|
private RecordingAdapter(String code, long sequenceId)
|
{
|
this.code = code;
|
this.sequenceId = sequenceId;
|
}
|
|
private List<String> calls()
|
{
|
return List.copyOf(calls);
|
}
|
|
@Override public String adapterCode() { return code; }
|
@Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); }
|
@Override public List<DetectedChannel> discoverChannels() { return List.of(); }
|
@Override public Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
|
{
|
return Map.of();
|
}
|
|
@Override public CabinetRealtime latestCabinet(ChannelIdentity channel)
|
{
|
calls.add(code);
|
return new CabinetRealtime(sequenceId, Instant.EPOCH, 0, 0, 0, 0, 0, 0, null, null, Instant.EPOCH,
|
null, "Asia/Hong_Kong", null);
|
}
|
|
@Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
|
{
|
return null;
|
}
|
|
@Override public List<StepRecord> stepRecords(ChannelIdentity channel) { return List.of(); }
|
@Override public List<CycleRecord> cycleRecords(ChannelIdentity channel) { return List.of(); }
|
|
@Override public long persistenceIntervalMillis(ChannelIdentity channel)
|
{
|
calls.add(code);
|
return "neware76".equals(code) ? 60000L : 70000L;
|
}
|
|
@Override public SampleWatermark currentWatermark(ChannelIdentity channel)
|
{
|
calls.add(code);
|
return new SampleWatermark(Instant.EPOCH, 1L);
|
}
|
|
@Override public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
|
{
|
calls.add(code);
|
return List.of();
|
}
|
}
|
}
|
