class LocalDeviceProductService extends GenericReactiveCrudService<DeviceProductEntity, String> {

    @Autowired
    private DeviceRegistry registry;

    @Autowired
    private ApplicationEventPublisher eventPublisher;

    @Autowired
    private ReactiveRepository<DeviceInstanceEntity, String> instanceRepository;


    public Mono<Integer> deploy(String id) {
        return findById(Mono.just(id))
            .doOnNext(this::validateDeviceProduct)
            .flatMap(product -> registry
                .register(product.toProductInfo())
                .onErrorMap(e -> new BusinessException("error.unable_to_load_protocol_by_access_id", 404, product.getMessageProtocol()))
                .then(
                    createUpdate()
                        .set(DeviceProductEntity::getState, DeviceProductState.registered.getValue())
                        .where(DeviceProductEntity::getId, id)
                        .execute()
                )
                .flatMap(i -> FastBeanCopier
                    .copy(product, new DeviceProductDeployEvent())
                    .publish(eventPublisher)
                    .thenReturn(i))
            );
    }

    private void validateDeviceProduct(DeviceProductEntity product) {
        // 设备接入ID不能为空
//        Assert.hasText(product.getAccessId(), "error.access_id_can_not_be_empty");
        // 发布前，必须填写消息协议
        Assert.hasText(product.getMessageProtocol(), "error.please_select_the_access_mode_first");
    }


    public Mono<Integer> cancelDeploy(String id) {
        return createUpdate()
            .set(DeviceProductEntity::getState, DeviceProductState.unregistered.getValue())
            .where(DeviceProductEntity::getId, id)
            .execute()
            .flatMap(integer ->
                         registry
                             .unregisterProduct(id)
                             .thenReturn(integer));

    }


    @Override
    public Mono<Integer> deleteById(Publisher<String> idPublisher) {<FILL_FUNCTION_BODY>}

}

        return Flux.from(idPublisher)
            .collectList()
            .flatMap(idList ->
                instanceRepository.createQuery()
                    .where()
                    .in(DeviceInstanceEntity::getProductId, idList)
                    .count()
                    .flatMap(i -> {
                        if (i > 0) {
                            return Mono.error(new IllegalArgumentException("存在关联设备,无法删除!"));
                        } else {
                            return super.deleteById(Flux.fromIterable(idList));
                        }
                    }));
    
<no_super_class>