lingwei.kong/hawkbit
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Move EventPublisher to hawkbit-repository-core making hawkbit-reposit… (#2576)

Browse Source 
* Move EventPublisher to hawkbit-repository-core making hawkbit-repository-api non dependent to rabbit binders

* Move EventPublisherHolder to org.eclipse.hawkbit.repository.event

* Fix overriding default values of RemoteEvents and rabbit binder

---------

Co-authored-by: vasilchev <vasil.ilchev@bosch.com>
This commit is contained in:
Vasil Ilchev
2025-08-01 08:48:07 +03:00
committed by GitHub
parent 107ebca740
commit cca105c80e
8 changed files with 42 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventPublisherConfiguration.java
View File

@@ -9,35 +9,58 @@
*/
package org.eclipse.hawkbit.event;
import java.io.IOException;
import java.util.concurrent.Executor;
import java.util.function.Consumer;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.repository.event.ApplicationEventFilter;
import org.eclipse.hawkbit.repository.event.EventPublisherHolder;
import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent;
import org.eclipse.hawkbit.repository.event.remote.RemoteTenantAwareEvent;
import org.eclipse.hawkbit.repository.event.EventPublisherHolder;
import org.eclipse.hawkbit.security.SystemSecurityContext;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.context.event.ApplicationEventMulticaster;
import org.springframework.context.event.SimpleApplicationEventMulticaster;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.io.support.ResourcePropertySource;
import org.springframework.messaging.converter.MessageConverter;
/**
* Autoconfiguration for the events.
*/
@Slf4j
@Configuration
@PropertySource("classpath:/hawkbit-events-defaults.properties")
public class EventPublisherConfiguration {
private final ConfigurableEnvironment environment;
public EventPublisherConfiguration(ConfigurableEnvironment environment) {
this.environment = environment;
}
@PostConstruct
public void registerProperties() {
try {
ResourcePropertySource props = new ResourcePropertySource("classpath:/hawkbit-events-defaults.properties");
// load manually to ensure that they are with the lowest precedence allowing to override them
environment.getPropertySources().addLast(props);
} catch (IOException ex) {
log.error("Failed to load default properties for event publisher", ex);
}
}
/**
* Server internal event publisher that allows parallel event processing if the event listener is marked as so.
*
@@ -116,29 +139,22 @@ public class EventPublisherConfiguration {
return publisher::publishEvent;
}
@Bean
@ConditionalOnProperty(name = "org.eclipse.hawkbit.events.remote-enabled", havingValue = "true")
@ConditionalOnProperty(name = "spring.cloud.stream.default.content-type", havingValue = "application/binary+stuff")
protected static class EventProtoStuffAutoConfiguration {
public MessageConverter eventMessageConverter(BindingProperties bindingProperties) {
/**
* @return the protostuff io message converter for events
*/
@Bean
public MessageConverter eventProtoStuffConverter() {
return new EventProtoStuffMessageConverter();
final String contentType = bindingProperties.getContentType();
if (contentType == null) {
throw new IllegalStateException("RemoteEvents are enabled and Content type must be specified in spring.cloud.stream.default.content-type.");
}
}
@ConditionalOnProperty(name = "org.eclipse.hawkbit.events.remote-enabled", havingValue = "true")
@ConditionalOnProperty(name = "spring.cloud.stream.default.content-type", havingValue = "application/remote-event-json")
protected static class EventJacksonAutoConfiguration {
/**
* @return the Jackson message converter for events
*/
@Bean
public MessageConverter eventJacksonMessageConverter() {
if (contentType.equals("application/binary+protostuff")) {
return new EventProtoStuffMessageConverter();
} else if (contentType.equals("application/remote-event-json")) {
return new EventJacksonMessageConverter();
} else {
throw new IllegalStateException("Unsupported content type: " + contentType + ". Supported types: application/x-protostuff, application/remote-event-json");
}
}
}

182
hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/repository/event/EventPublisherHolder.java Normal file
View File

@@ -0,0 +1,182 @@
/**
* Copyright (c) 2025 Contributors to the Eclipse Foundation
*
* This program and the accompanying materials are made
* available under the terms of the Eclipse Public License 2.0
* which is available at https://www.eclipse.org/legal/epl-2.0/
*
* SPDX-License-Identifier: EPL-2.0
*/
package org.eclipse.hawkbit.repository.event;
import java.util.Set;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent;
import org.eclipse.hawkbit.repository.event.remote.CancelTargetAssignmentEvent;
import org.eclipse.hawkbit.repository.event.remote.MultiActionAssignEvent;
import org.eclipse.hawkbit.repository.event.remote.MultiActionCancelEvent;
import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.event.remote.TargetAttributesRequestedEvent;
import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent;
import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent;
import org.eclipse.hawkbit.repository.event.remote.service.CancelTargetAssignmentServiceEvent;
import org.eclipse.hawkbit.repository.event.remote.service.MultiActionAssignServiceEvent;
import org.eclipse.hawkbit.repository.event.remote.service.MultiActionCancelServiceEvent;
import org.eclipse.hawkbit.repository.event.remote.service.TargetAssignDistributionSetServiceEvent;
import org.eclipse.hawkbit.repository.event.remote.service.TargetAttributesRequestedServiceEvent;
import org.eclipse.hawkbit.repository.event.remote.service.TargetCreatedServiceEvent;
import org.eclipse.hawkbit.repository.event.remote.service.TargetDeletedServiceEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
@Slf4j
public final class EventPublisherHolder {
@Value("${org.eclipse.hawkbit.events.remote-enabled:true}")
private boolean remoteEventsEnabled;
@Value("${org.eclipse.hawkbit.events.remote.destination:fanoutEventChannel}")
private String fanoutEventChannel;
@Value("${org.eclipse.hawkbit.events.remote-service-enabled:true}")
private boolean remoteServiceEventsEnabled;
@Value("${org.eclipse.hawkbit.events.remote.service.destination:serviceEventChannel}")
private String serviceEventChannel;
private static final EventPublisherHolder SINGLETON = new EventPublisherHolder();
private ApplicationEventPublisher delegateEventPublisher;
private StreamBridge streamBridge;
public static EventPublisherHolder getInstance() {
return SINGLETON;
}
@PostConstruct
private void validateRemoteEventConfig() {
if (remoteEventsEnabled && streamBridge == null) {
throw new IllegalStateException("'org.eclipse.hawkbit.events.remote-enabled' is true but streamBridge is not configured. Check if 'spring-cloud-starter-stream-rabbit' dependency is included.");
}
}
public static final Set<Class<?>> SERVICE_EVENTS = Set.of(
TargetCreatedEvent.class,
TargetDeletedEvent.class,
MultiActionAssignEvent.class,
MultiActionCancelEvent.class,
TargetAssignDistributionSetEvent.class,
TargetAttributesRequestedEvent.class,
CancelTargetAssignmentEvent.class
);
@Autowired
public void setApplicationEventPublisher(final ApplicationEventPublisher delegate) {
this.delegateEventPublisher = delegate;
}
@Autowired(required = false)
public void setStreamBridge(final StreamBridge streamBridge) {
this.streamBridge = streamBridge;
}
public ApplicationEventPublisher getEventPublisher() {
return new RoutingEventPublisher(streamBridge, delegateEventPublisher);
}
class RoutingEventPublisher implements ApplicationEventPublisher {
private final StreamBridge streamBridge;
private final ApplicationEventPublisher delegate;
public RoutingEventPublisher(final StreamBridge streamBridge, final ApplicationEventPublisher delegate) {
this.streamBridge = streamBridge;
this.delegate = delegate;
}
@Override
public void publishEvent(final Object event) {
routeEvent(event);
}
@Override
public void publishEvent(final ApplicationEvent event) {
routeEvent(event);
}
private void routeEvent(Object event) {
if (remoteEventsEnabled && event instanceof AbstractRemoteEvent remoteEvent) {
// send events to remote nodes
publishRemotely(remoteEvent);
} else {
// publish locally
publishLocally(event);
}
}
private void publishRemotely(final AbstractRemoteEvent remoteEvent) {
streamBridge.send(fanoutEventChannel, remoteEvent);
// some events need to be processed only by single service replica
// wrap the entity event into a service event and send it to the service channel
if (shouldForwardAsServiceEvent(remoteEvent)) {
final AbstractRemoteEvent serviceEvent = toServiceEvent(remoteEvent);
if (serviceEvent != null) {
log.debug("Publishing Service event: {} to remote channel: {}", serviceEvent, serviceEventChannel);
streamBridge.send(serviceEventChannel, serviceEvent);
} else {
log.error("No Service event created for: {}. Skipping send Service event to Service channel. {}", remoteEvent.getClass(),
serviceEventChannel);
}
}
}
private void publishLocally(final Object event) {
delegate.publishEvent(event);
// check if the event should be forwarded as a service event even if it is not a remote event
if (shouldForwardAsServiceEvent(event)) {
final AbstractRemoteEvent serviceEvent = toServiceEvent((AbstractRemoteEvent) event);
if (serviceEvent != null) {
log.debug("Publishing Service event: {} to locally.", serviceEvent);
delegate.publishEvent(serviceEvent);
} else {
log.error("No Service event created for: {}. Skipping send Service event locally.", event.getClass());
}
}
}
/**
* Checks if the event should be forwarded as a service event.
* If remote service events are enabled and the event is one of the service events,
*
* @param remoteEvent the event to check whether it should be forwarded as a service event
* @return true if the event should be forwarded as a service event, false otherwise
*/
private boolean shouldForwardAsServiceEvent(final Object remoteEvent) {
return remoteServiceEventsEnabled && SERVICE_EVENTS.contains(remoteEvent.getClass());
}
private AbstractRemoteEvent toServiceEvent(final AbstractRemoteEvent event) {
if (event instanceof TargetAssignDistributionSetEvent targetAssignDistributionSetEvent) {
return new TargetAssignDistributionSetServiceEvent(targetAssignDistributionSetEvent);
} else if (event instanceof MultiActionAssignEvent multiActionAssignEvent) {
return new MultiActionAssignServiceEvent(multiActionAssignEvent);
} else if (event instanceof MultiActionCancelEvent multiActionCancelEvent) {
return new MultiActionCancelServiceEvent(multiActionCancelEvent);
} else if (event instanceof CancelTargetAssignmentEvent cancelTargetAssignmentEvent) {
return new CancelTargetAssignmentServiceEvent(cancelTargetAssignmentEvent);
} else if (event instanceof TargetDeletedEvent targetDeletedEvent) {
return new TargetDeletedServiceEvent(targetDeletedEvent);
} else if (event instanceof TargetCreatedEvent targetCreatedEvent) {
return new TargetCreatedServiceEvent(targetCreatedEvent);
} else if (event instanceof TargetAttributesRequestedEvent targetAttributesRequestedEvent) {
return new TargetAttributesRequestedServiceEvent(targetAttributesRequestedEvent);
}
return null;
}
}
}