Feature horizontal scalability (#305)
Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com>
This commit is contained in:
Dennis Melzer
committed by
Kai Zimmermann
Kai Zimmermann
parent
07cb62a3dd
commit
866bc72114
@@ -0,0 +1,108 @@
|
||||
/**
|
||||
* Copyright (c) 2015 Bosch Software Innovations GmbH and others.
|
||||
*
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* which accompanies this distribution, and is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*/
|
||||
package org.eclipse.hawkbit.event;
|
||||
|
||||
import org.apache.commons.lang3.ClassUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
|
||||
import org.springframework.integration.support.MutableMessageHeaders;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.converter.AbstractMessageConverter;
|
||||
import org.springframework.messaging.converter.MessageConversionException;
|
||||
import org.springframework.messaging.support.MessageHeaderAccessor;
|
||||
import org.springframework.util.MimeType;
|
||||
|
||||
import io.protostuff.LinkedBuffer;
|
||||
import io.protostuff.ProtobufIOUtil;
|
||||
import io.protostuff.ProtostuffIOUtil;
|
||||
import io.protostuff.Schema;
|
||||
import io.protostuff.runtime.RuntimeSchema;
|
||||
|
||||
/**
|
||||
* A customize message converter for the spring cloud events. The converter is
|
||||
* registered for the application/binary+protostuff type.
|
||||
*
|
||||
*/
|
||||
public class BusProtoStuffMessageConverter extends AbstractMessageConverter {
|
||||
|
||||
public static final MimeType APPLICATION_BINARY_PROTOSTUFF = new MimeType("application", "binary+protostuff");
|
||||
private static final Logger LOG = LoggerFactory.getLogger(BusProtoStuffMessageConverter.class);
|
||||
private static final String DEFAULT_CLASS_FIELD_NAME = "__Class__";
|
||||
|
||||
/**
|
||||
* Constructor.
|
||||
*/
|
||||
public BusProtoStuffMessageConverter() {
|
||||
super(APPLICATION_BINARY_PROTOSTUFF);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean supports(final Class<?> aClass) {
|
||||
return RemoteApplicationEvent.class.isAssignableFrom(aClass);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object convertFromInternal(final Message<?> message, final Class<?> targetClass,
|
||||
final Object conversionHint) {
|
||||
final Object payload = message.getPayload();
|
||||
|
||||
try {
|
||||
final Class<?> deserializeClass = ClassUtils
|
||||
.getClass(message.getHeaders().get(DEFAULT_CLASS_FIELD_NAME).toString());
|
||||
if (payload instanceof byte[]) {
|
||||
@SuppressWarnings("unchecked")
|
||||
final Schema<Object> schema = (Schema<Object>) RuntimeSchema.getSchema(deserializeClass);
|
||||
final Object deserializeEvent = schema.newMessage();
|
||||
ProtobufIOUtil.mergeFrom((byte[]) message.getPayload(), deserializeEvent, schema);
|
||||
return deserializeEvent;
|
||||
}
|
||||
} catch (final ClassNotFoundException e) {
|
||||
LOG.error("Protostuff cannot find derserialize class", e);
|
||||
throw new MessageConversionException(message, "Failed to read payload", e);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Object convertToInternal(final Object payload, final MessageHeaders headers,
|
||||
final Object conversionHint) {
|
||||
checkIfHeaderMutable(headers);
|
||||
final Class<? extends Object> serializeClass = payload.getClass();
|
||||
@SuppressWarnings("unchecked")
|
||||
final Schema<Object> schema = (Schema<Object>) RuntimeSchema.getSchema(serializeClass);
|
||||
final LinkedBuffer buffer = LinkedBuffer.allocate();
|
||||
final byte[] serializeByte;
|
||||
try {
|
||||
serializeByte = ProtostuffIOUtil.toByteArray(payload, schema, buffer);
|
||||
} finally {
|
||||
buffer.clear();
|
||||
}
|
||||
|
||||
headers.put(DEFAULT_CLASS_FIELD_NAME, serializeClass.getName());
|
||||
return serializeByte;
|
||||
}
|
||||
|
||||
private static void checkIfHeaderMutable(final MessageHeaders headers) {
|
||||
if (isAccessorMutable(headers) || headers instanceof MutableMessageHeaders) {
|
||||
return;
|
||||
}
|
||||
LOG.error("Protostuff cannot set serializae class because message header is not mutable");
|
||||
throw new MessageConversionException(
|
||||
"Cannot set the serialize class to message header. Need Mutable message header");
|
||||
}
|
||||
|
||||
private static boolean isAccessorMutable(final MessageHeaders headers) {
|
||||
final MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(headers, MessageHeaderAccessor.class);
|
||||
return accessor != null && accessor.isMutable();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -6,13 +6,13 @@
|
||||
* which accompanies this distribution, and is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*/
|
||||
package org.eclipse.hawkbit.repository.jpa;
|
||||
package org.eclipse.hawkbit.repository;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
|
||||
import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.helper.TenantConfigurationManagementHolder;
|
||||
import org.eclipse.hawkbit.repository.model.helper.TenantConfigurationManagementHolder;
|
||||
import org.eclipse.hawkbit.tenancy.configuration.DurationHelper;
|
||||
import org.eclipse.hawkbit.tenancy.configuration.TenantConfigurationKey;
|
||||
|
||||
@@ -1,54 +0,0 @@
|
||||
/**
|
||||
* Copyright (c) 2015 Bosch Software Innovations GmbH and others.
|
||||
*
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* which accompanies this distribution, and is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*/
|
||||
package org.eclipse.hawkbit.repository.jpa.model.helper;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
|
||||
import com.google.common.eventbus.EventBus;
|
||||
|
||||
/**
|
||||
* A singleton bean which holds the {@link EventBus} to have to the cache
|
||||
* manager in beans not instantiated by spring e.g. JPA entities or
|
||||
* CacheFieldEntityListener which cannot be autowired.
|
||||
*
|
||||
*/
|
||||
public final class EventBusHolder {
|
||||
|
||||
private static final EventBusHolder SINGLETON = new EventBusHolder();
|
||||
|
||||
@Autowired
|
||||
private EventBus eventBus;
|
||||
|
||||
private EventBusHolder() {
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the cache manager holder singleton instance
|
||||
*/
|
||||
public static EventBusHolder getInstance() {
|
||||
return SINGLETON;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the eventBus
|
||||
*/
|
||||
public EventBus getEventBus() {
|
||||
return eventBus;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param eventBus
|
||||
* the eventBus to set
|
||||
*/
|
||||
public void setEventBus(final EventBus eventBus) {
|
||||
this.eventBus = eventBus;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
/**
|
||||
* Copyright (c) 2015 Bosch Software Innovations GmbH and others.
|
||||
*
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* which accompanies this distribution, and is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*/
|
||||
package org.eclipse.hawkbit.repository.model.helper;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
|
||||
/**
|
||||
* A singleton bean which holds the event publisher to have to the cache manager
|
||||
* in beans not instantiated by spring e.g. JPA entities or
|
||||
* CacheFieldEntityListener which cannot be autowired.
|
||||
*
|
||||
*/
|
||||
public final class EventPublisherHolder {
|
||||
|
||||
private static final EventPublisherHolder SINGLETON = new EventPublisherHolder();
|
||||
|
||||
@Autowired
|
||||
private ApplicationEventPublisher eventPublisher;
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext applicationContext;
|
||||
|
||||
private EventPublisherHolder() {
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the cache manager holder singleton instance
|
||||
*/
|
||||
public static EventPublisherHolder getInstance() {
|
||||
return SINGLETON;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the eventPublisher
|
||||
*/
|
||||
public ApplicationEventPublisher getEventPublisher() {
|
||||
return eventPublisher;
|
||||
}
|
||||
|
||||
public String getApplicationId() {
|
||||
return applicationContext.getId();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -6,7 +6,7 @@
|
||||
* which accompanies this distribution, and is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*/
|
||||
package org.eclipse.hawkbit.repository.jpa.model.helper;
|
||||
package org.eclipse.hawkbit.repository.model.helper;
|
||||
|
||||
import org.eclipse.hawkbit.repository.SystemManagement;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
@@ -6,7 +6,7 @@
|
||||
* which accompanies this distribution, and is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*/
|
||||
package org.eclipse.hawkbit.repository.jpa.model.helper;
|
||||
package org.eclipse.hawkbit.repository.model.helper;
|
||||
|
||||
import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
@@ -6,19 +6,14 @@
|
||||
* which accompanies this distribution, and is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*/
|
||||
package org.eclipse.hawkbit.repository.jpa.rsql;
|
||||
package org.eclipse.hawkbit.repository.rsql;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
|
||||
import org.apache.commons.lang3.text.StrLookup;
|
||||
import org.apache.commons.lang3.text.StrSubstitutor;
|
||||
import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
|
||||
import org.eclipse.hawkbit.repository.jpa.TimestampCalculator;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.helper.TenantConfigurationManagementHolder;
|
||||
import org.eclipse.hawkbit.repository.TimestampCalculator;
|
||||
import org.eclipse.hawkbit.repository.rsql.VirtualPropertyReplacer;
|
||||
import org.eclipse.hawkbit.tenancy.configuration.DurationHelper;
|
||||
import org.eclipse.hawkbit.tenancy.configuration.TenantConfigurationKey;
|
||||
|
||||
/**
|
||||
* Adds macro capabilities to RSQL expressions that are used to filter for
|
||||
Reference in New Issue
Block a user