package com.battery.device.api.registry;
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertFalse;
|
import static org.junit.jupiter.api.Assertions.assertSame;
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
import static org.junit.jupiter.api.Assertions.assertNotSame;
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.concurrent.CountDownLatch;
|
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.Future;
|
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.atomic.AtomicBoolean;
|
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.DeviceControlCommand;
|
|
class DeviceAdapterRegistryTest
|
{
|
@Test
|
void synchronizesEnabledConfigurationWithoutRecreatingUnchangedAdapter()
|
{
|
RecordingFactory factory = new RecordingFactory();
|
try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of(factory)))
|
{
|
AdapterInstanceConfig config = config("127.0.0.1", true);
|
|
registry.synchronize(config);
|
DeviceAdapter first = registry.execute("lab-bts", adapter -> adapter);
|
registry.synchronize(config);
|
|
assertSame(first, registry.execute("lab-bts", adapter -> adapter));
|
assertFalse(factory.adapters.get(0).closed.get());
|
}
|
}
|
|
@Test
|
void replacesChangedConfigurationAndClosesAdapterWhenDisabled()
|
{
|
RecordingFactory factory = new RecordingFactory();
|
try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of(factory)))
|
{
|
registry.synchronize(config("127.0.0.1", true));
|
StubAdapter first = factory.adapters.get(0);
|
|
registry.synchronize(config("127.0.0.2", true));
|
|
assertTrue(first.closed.get());
|
assertNotSame(first, registry.execute("lab-bts", adapter -> adapter));
|
|
registry.synchronize(config("127.0.0.2", false));
|
|
assertTrue(factory.adapters.get(1).closed.get());
|
assertFalse(registry.contains("lab-bts"));
|
assertThrows(AdapterRegistryException.class,
|
() -> registry.execute("lab-bts", adapter -> adapter));
|
}
|
}
|
|
@Test
|
void serializesOperationsForTheSameAdapterInstance() throws Exception
|
{
|
RecordingFactory factory = new RecordingFactory();
|
try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of(factory)))
|
{
|
registry.synchronize(config("127.0.0.1", true));
|
CountDownLatch firstEntered = new CountDownLatch(1);
|
CountDownLatch releaseFirst = new CountDownLatch(1);
|
CountDownLatch secondEntered = new CountDownLatch(1);
|
ExecutorService executor = Executors.newFixedThreadPool(2);
|
try
|
{
|
Future<?> first = executor.submit(() -> registry.execute("lab-bts", adapter -> {
|
firstEntered.countDown();
|
await(releaseFirst);
|
return null;
|
}));
|
assertTrue(firstEntered.await(1, TimeUnit.SECONDS));
|
|
Future<?> second = executor.submit(() -> registry.execute("lab-bts", adapter -> {
|
secondEntered.countDown();
|
return null;
|
}));
|
|
assertFalse(secondEntered.await(100, TimeUnit.MILLISECONDS));
|
releaseFirst.countDown();
|
first.get(1, TimeUnit.SECONDS);
|
second.get(1, TimeUnit.SECONDS);
|
assertTrue(secondEntered.await(1, TimeUnit.SECONDS));
|
}
|
finally
|
{
|
executor.shutdownNow();
|
}
|
}
|
}
|
|
@Test
|
void configurationReplacementWaitsForRunningOperationBeforeClosingAdapter() throws Exception
|
{
|
RecordingFactory factory = new RecordingFactory();
|
try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of(factory)))
|
{
|
registry.synchronize(config("127.0.0.1", true));
|
StubAdapter firstAdapter = factory.adapters.get(0);
|
CountDownLatch operationEntered = new CountDownLatch(1);
|
CountDownLatch releaseOperation = new CountDownLatch(1);
|
CountDownLatch replacementStarted = new CountDownLatch(1);
|
ExecutorService executor = Executors.newFixedThreadPool(2);
|
try
|
{
|
Future<?> operation = executor.submit(() -> registry.execute("lab-bts", adapter -> {
|
operationEntered.countDown();
|
await(releaseOperation);
|
return null;
|
}));
|
assertTrue(operationEntered.await(1, TimeUnit.SECONDS));
|
|
Future<?> replacement = executor.submit(() -> {
|
replacementStarted.countDown();
|
registry.synchronize(config("127.0.0.2", true));
|
});
|
assertTrue(replacementStarted.await(1, TimeUnit.SECONDS));
|
assertFalse(replacement.isDone());
|
assertFalse(firstAdapter.closed.get());
|
|
releaseOperation.countDown();
|
operation.get(1, TimeUnit.SECONDS);
|
replacement.get(1, TimeUnit.SECONDS);
|
assertTrue(firstAdapter.closed.get());
|
}
|
finally
|
{
|
executor.shutdownNow();
|
}
|
}
|
}
|
|
private static void await(CountDownLatch latch)
|
{
|
try
|
{
|
if (!latch.await(1, TimeUnit.SECONDS)) throw new AssertionError("等待适配器测试信号超时");
|
}
|
catch (InterruptedException exception)
|
{
|
Thread.currentThread().interrupt();
|
throw new AssertionError("等待适配器测试信号被中断", exception);
|
}
|
}
|
|
/**
|
* 多来源实例：取数按适配器编码解析到各自实例；实例缺失时明确失败，不回退到其他实例。
|
*/
|
@Test
|
void routesTestDataObservationToTheInstanceNamedByTheChannelAndFailsWithoutFallback()
|
{
|
try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of(new ObservableFactory())))
|
{
|
registry.register(observableConfig("source-a", 1000), new ObservableAdapter("source-a", 1000L));
|
registry.register(observableConfig("source-b", 2000), new ObservableAdapter("source-b", 2000L));
|
|
assertEquals(1000L, registry.executeTestData("source-a",
|
adapter -> adapter.latestCabinet(channel("source-a")).sequenceId()).longValue());
|
assertEquals(2000L, registry.executeTestData("source-b",
|
adapter -> adapter.latestCabinet(channel("source-b")).sequenceId()).longValue());
|
|
// 实例缺失或未提供扩展观测契约时明确失败，不能落到别的实例上。
|
assertThrows(AdapterRegistryException.class,
|
() -> registry.executeTestData("source-c", adapter -> adapter.latestCabinet(channel("source-a"))));
|
assertThrows(AdapterRegistryException.class,
|
() -> registry.executeTestData("lab-bts", adapter -> adapter.adapterCode()));
|
}
|
}
|
|
/** 代码装配的实例与工厂实例共用同一份实例表，发现与取数因此按同一编码解析。 */
|
@Test
|
void codeAssembledInstanceIsVisibleToBothAccessors()
|
{
|
try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of()))
|
{
|
ObservableAdapter assembled = new ObservableAdapter("assembled", 42L);
|
registry.register(observableConfig("assembled", 3000), assembled);
|
|
assertSame(assembled, registry.execute("assembled", adapter -> adapter));
|
assertSame(assembled, registry.executeTestData("assembled", adapter -> adapter));
|
}
|
}
|
|
private static ChannelIdentity channel(String adapterCode)
|
{
|
return new ChannelIdentity(adapterCode, 22, 1, 1, 1);
|
}
|
|
private static AdapterInstanceConfig observableConfig(String adapterCode, int pollIntervalMs)
|
{
|
return new AdapterInstanceConfig(adapterCode, "NEWARE_MYSQL", "127.0.0.1", 3306, pollIntervalMs,
|
"Asia/Hong_Kong", true, AdapterCapabilities.observeOnly());
|
}
|
|
private static final class ObservableFactory implements DeviceAdapterFactory
|
{
|
@Override
|
public String adapterType()
|
{
|
return "NEWARE_MYSQL";
|
}
|
|
@Override
|
public DeviceAdapter create(AdapterInstanceConfig config)
|
{
|
return new ObservableAdapter(config.adapterCode(), 0L);
|
}
|
}
|
|
/** 同时实现发现与测试数据观测契约的最小实例，用于验证按编码路由。 */
|
private static final class ObservableAdapter implements DeviceAdapter, TestDataObservationAdapter
|
{
|
private final String code;
|
private final long sequenceId;
|
|
private ObservableAdapter(String code, long sequenceId)
|
{
|
this.code = code;
|
this.sequenceId = sequenceId;
|
}
|
|
@Override
|
public String adapterCode()
|
{
|
return code;
|
}
|
|
@Override
|
public AdapterCapabilities capabilities()
|
{
|
return AdapterCapabilities.observeOnly();
|
}
|
|
@Override
|
public List<com.battery.device.api.model.DetectedChannel> discoverChannels()
|
{
|
return List.of();
|
}
|
|
@Override
|
public java.util.Map<ChannelIdentity, com.battery.device.api.model.RealtimeSnapshot> inquire(
|
java.util.Set<ChannelIdentity> channels)
|
{
|
return java.util.Map.of();
|
}
|
|
@Override
|
public TestDataObservationAdapter.CabinetRealtime latestCabinet(ChannelIdentity channel)
|
{
|
return new TestDataObservationAdapter.CabinetRealtime(sequenceId, java.time.Instant.EPOCH, 0, 0, 0, 0,
|
0, 0, null, null, java.time.Instant.EPOCH, null, "Asia/Hong_Kong", null);
|
}
|
|
@Override
|
public TestDataObservationAdapter.CollectorRealtime latestCollector(String adapterCode, String collectorCode)
|
{
|
return null;
|
}
|
|
@Override
|
public List<TestDataObservationAdapter.StepRecord> stepRecords(ChannelIdentity channel)
|
{
|
return List.of();
|
}
|
|
@Override
|
public List<TestDataObservationAdapter.CycleRecord> cycleRecords(ChannelIdentity channel)
|
{
|
return List.of();
|
}
|
|
@Override
|
public long persistenceIntervalMillis(ChannelIdentity channel)
|
{
|
return 60000L;
|
}
|
}
|
|
private static AdapterInstanceConfig config(String host, boolean enabled)
|
{
|
return new AdapterInstanceConfig("lab-bts", "BTS", host, 502, 5000,
|
"Asia/Hong_Kong", enabled, AdapterCapabilities.observeOnly());
|
}
|
|
private static final class RecordingFactory implements DeviceAdapterFactory
|
{
|
private final List<StubAdapter> adapters = new ArrayList<>();
|
|
@Override
|
public String adapterType()
|
{
|
return "BTS";
|
}
|
|
@Override
|
public DeviceAdapter create(AdapterInstanceConfig config)
|
{
|
StubAdapter adapter = new StubAdapter(config.adapterCode());
|
adapters.add(adapter);
|
return adapter;
|
}
|
}
|
|
private static final class StubAdapter implements DeviceAdapter
|
{
|
private final String code;
|
private final AtomicBoolean closed = new AtomicBoolean();
|
|
private StubAdapter(String code)
|
{
|
this.code = code;
|
}
|
|
@Override
|
public String adapterCode()
|
{
|
return code;
|
}
|
|
@Override
|
public AdapterCapabilities capabilities()
|
{
|
return AdapterCapabilities.observeOnly();
|
}
|
|
@Override
|
public List<com.battery.device.api.model.DetectedChannel> discoverChannels()
|
{
|
return List.of();
|
}
|
|
@Override
|
public java.util.Map<ChannelIdentity, com.battery.device.api.model.RealtimeSnapshot> inquire(
|
java.util.Set<ChannelIdentity> channels)
|
{
|
return java.util.Map.of();
|
}
|
|
@Override
|
public void control(ChannelIdentity channel, DeviceControlCommand command)
|
{
|
throw new UnsupportedOperationException();
|
}
|
|
@Override
|
public void close()
|
{
|
closed.set(true);
|
}
|
}
|
}
|
