Browse Source
* Created kafka and in-memory queue providers * Added header requestTime * refactoredpull/2566/head
Yevhen Bondarenko
6 years ago
committed by
GitHub
GitHub
50 changed files with 905 additions and 692 deletions
@ -1,15 +1,28 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
|
|||
public interface TbQueueProducer<T extends TbQueueMsg> { |
|||
|
|||
void init(); |
|||
|
|||
String getDefaultTopic(); |
|||
|
|||
ListenableFuture<TbQueueMsgMetadata> send(T msg, TbQueueCallback callback); |
|||
void send(T msg, TbQueueCallback callback); |
|||
|
|||
ListenableFuture<TbQueueMsgMetadata> send(String topic, T msg, TbQueueCallback callback); |
|||
void send(String topic, T msg, TbQueueCallback callback); |
|||
|
|||
} |
|||
|
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server; |
|||
|
|||
import lombok.Data; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
@Data |
|||
@Component |
|||
public class TbQueueTransportApiSettings { |
|||
@Value("${queue.transport_api.requests_topic}") |
|||
private String requestsTopic; |
|||
|
|||
@Value("${queue.transport_api.responses_topic}") |
|||
private String responsesTopic; |
|||
|
|||
@Value("${queue.transport_api.max_pending_requests}") |
|||
private int maxPendingRequests; |
|||
|
|||
@Value("${queue.transport_api.max_requests_timeout}") |
|||
private int maxRequestsTimeout; |
|||
|
|||
@Value("${queue.transport_api.response_poll_interval}") |
|||
private long responsePollInterval; |
|||
|
|||
} |
|||
@ -1,50 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.nio.ByteBuffer; |
|||
import java.nio.charset.StandardCharsets; |
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* Created by ashvayka on 25.09.18. |
|||
*/ |
|||
@Slf4j |
|||
public abstract class AbstractTbKafkaTemplate { |
|||
protected byte[] uuidToBytes(UUID uuid) { |
|||
ByteBuffer buf = ByteBuffer.allocate(16); |
|||
buf.putLong(uuid.getMostSignificantBits()); |
|||
buf.putLong(uuid.getLeastSignificantBits()); |
|||
return buf.array(); |
|||
} |
|||
|
|||
protected static UUID bytesToUuid(byte[] bytes) { |
|||
ByteBuffer bb = ByteBuffer.wrap(bytes); |
|||
long firstLong = bb.getLong(); |
|||
long secondLong = bb.getLong(); |
|||
return new UUID(firstLong, secondLong); |
|||
} |
|||
|
|||
protected byte[] stringToBytes(String string) { |
|||
return string.getBytes(StandardCharsets.UTF_8); |
|||
} |
|||
|
|||
protected String bytesToString(byte[] data) { |
|||
return new String(data, StandardCharsets.UTF_8); |
|||
} |
|||
} |
|||
@ -1,66 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
|
|||
import java.util.concurrent.Executor; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public class AsyncCallbackTemplate { |
|||
|
|||
public static <T> void withCallbackAndTimeout(ListenableFuture<T> future, |
|||
Consumer<T> onSuccess, |
|||
Consumer<Throwable> onFailure, |
|||
long timeoutInMs, |
|||
ScheduledExecutorService timeoutExecutor, |
|||
Executor callbackExecutor) { |
|||
future = Futures.withTimeout(future, timeoutInMs, TimeUnit.MILLISECONDS, timeoutExecutor); |
|||
withCallback(future, onSuccess, onFailure, callbackExecutor); |
|||
} |
|||
|
|||
public static <T> void withCallback(ListenableFuture<T> future, Consumer<T> onSuccess, |
|||
Consumer<Throwable> onFailure, Executor executor) { |
|||
FutureCallback<T> callback = new FutureCallback<T>() { |
|||
@Override |
|||
public void onSuccess(T result) { |
|||
try { |
|||
onSuccess.accept(result); |
|||
} catch (Throwable th) { |
|||
onFailure(th); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
onFailure.accept(t); |
|||
} |
|||
}; |
|||
if (executor != null) { |
|||
Futures.addCallback(future, callback, executor); |
|||
} else { |
|||
Futures.addCallback(future, callback); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,213 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.SettableFuture; |
|||
import lombok.Builder; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.kafka.clients.admin.CreateTopicsResult; |
|||
import org.apache.kafka.clients.admin.NewTopic; |
|||
import org.apache.kafka.clients.consumer.ConsumerRecords; |
|||
import org.apache.kafka.clients.producer.Callback; |
|||
import org.apache.kafka.clients.producer.RecordMetadata; |
|||
import org.apache.kafka.common.errors.InterruptException; |
|||
import org.apache.kafka.common.errors.TopicExistsException; |
|||
import org.apache.kafka.common.header.Header; |
|||
import org.apache.kafka.common.header.internals.RecordHeader; |
|||
|
|||
import java.io.IOException; |
|||
import java.time.Duration; |
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeoutException; |
|||
|
|||
/** |
|||
* Created by ashvayka on 25.09.18. |
|||
*/ |
|||
@Slf4j |
|||
public class TbKafkaRequestTemplate<Request, Response> extends AbstractTbKafkaTemplate { |
|||
|
|||
private final TBKafkaProducerTemplate<Request> requestTemplate; |
|||
private final TBKafkaConsumerTemplate<Response> responseTemplate; |
|||
private final ConcurrentMap<UUID, ResponseMetaData<Response>> pendingRequests; |
|||
private final boolean internalExecutor; |
|||
private final ExecutorService executor; |
|||
private final long maxRequestTimeout; |
|||
private final long maxPendingRequests; |
|||
private final long pollInterval; |
|||
private volatile long tickTs = 0L; |
|||
private volatile long tickSize = 0L; |
|||
private volatile boolean stopped = false; |
|||
|
|||
@Builder |
|||
public TbKafkaRequestTemplate(TBKafkaProducerTemplate<Request> requestTemplate, |
|||
TBKafkaConsumerTemplate<Response> responseTemplate, |
|||
long maxRequestTimeout, |
|||
long maxPendingRequests, |
|||
long pollInterval, |
|||
ExecutorService executor) { |
|||
this.requestTemplate = requestTemplate; |
|||
this.responseTemplate = responseTemplate; |
|||
this.pendingRequests = new ConcurrentHashMap<>(); |
|||
this.maxRequestTimeout = maxRequestTimeout; |
|||
this.maxPendingRequests = maxPendingRequests; |
|||
this.pollInterval = pollInterval; |
|||
if (executor != null) { |
|||
internalExecutor = false; |
|||
this.executor = executor; |
|||
} else { |
|||
internalExecutor = true; |
|||
this.executor = Executors.newSingleThreadExecutor(); |
|||
} |
|||
} |
|||
|
|||
public void init() { |
|||
try { |
|||
TBKafkaAdmin admin = new TBKafkaAdmin(this.requestTemplate.getSettings()); |
|||
CreateTopicsResult result = admin.createTopic(new NewTopic(responseTemplate.getTopic(), 1, (short) 1)); |
|||
result.all().get(); |
|||
} catch (Exception e) { |
|||
if ((e instanceof TopicExistsException) || (e.getCause() != null && e.getCause() instanceof TopicExistsException)) { |
|||
log.trace("[{}] Topic already exists. ", responseTemplate.getTopic()); |
|||
} else { |
|||
log.info("[{}] Failed to create topic: {}", responseTemplate.getTopic(), e.getMessage(), e); |
|||
throw new RuntimeException(e); |
|||
} |
|||
|
|||
} |
|||
this.requestTemplate.init(); |
|||
tickTs = System.currentTimeMillis(); |
|||
responseTemplate.subscribe(); |
|||
executor.submit(() -> { |
|||
long nextCleanupMs = 0L; |
|||
while (!stopped) { |
|||
try { |
|||
ConsumerRecords<String, byte[]> responses = responseTemplate.poll(Duration.ofMillis(pollInterval)); |
|||
if (responses.count() > 0) { |
|||
log.trace("Polling responses completed, consumer records count [{}]", responses.count()); |
|||
} |
|||
responses.forEach(response -> { |
|||
log.trace("Received response to Kafka Template request: {}", response); |
|||
Header requestIdHeader = response.headers().lastHeader(TbKafkaSettings.REQUEST_ID_HEADER); |
|||
Response decodedResponse = null; |
|||
UUID requestId = null; |
|||
if (requestIdHeader == null) { |
|||
try { |
|||
decodedResponse = responseTemplate.decode(response); |
|||
requestId = responseTemplate.extractRequestId(decodedResponse); |
|||
} catch (IOException e) { |
|||
log.error("Failed to decode response", e); |
|||
} |
|||
} else { |
|||
requestId = bytesToUuid(requestIdHeader.value()); |
|||
} |
|||
if (requestId == null) { |
|||
log.error("[{}] Missing requestId in header and body", response); |
|||
} else { |
|||
log.trace("[{}] Response received", requestId); |
|||
ResponseMetaData<Response> expectedResponse = pendingRequests.remove(requestId); |
|||
if (expectedResponse == null) { |
|||
log.trace("[{}] Invalid or stale request", requestId); |
|||
} else { |
|||
try { |
|||
if (decodedResponse == null) { |
|||
decodedResponse = responseTemplate.decode(response); |
|||
} |
|||
expectedResponse.future.set(decodedResponse); |
|||
} catch (IOException e) { |
|||
expectedResponse.future.setException(e); |
|||
} |
|||
} |
|||
} |
|||
}); |
|||
tickTs = System.currentTimeMillis(); |
|||
tickSize = pendingRequests.size(); |
|||
if (nextCleanupMs < tickTs) { |
|||
//cleanup;
|
|||
pendingRequests.forEach((key, value) -> { |
|||
if (value.expTime < tickTs) { |
|||
ResponseMetaData<Response> staleRequest = pendingRequests.remove(key); |
|||
if (staleRequest != null) { |
|||
log.trace("[{}] Request timeout detected, expTime [{}], tickTs [{}]", key, staleRequest.expTime, tickTs); |
|||
staleRequest.future.setException(new TimeoutException()); |
|||
} |
|||
} |
|||
}); |
|||
nextCleanupMs = tickTs + maxRequestTimeout; |
|||
} |
|||
} catch (InterruptException ie) { |
|||
if (!stopped) { |
|||
log.warn("Fetching data from kafka was interrupted.", ie); |
|||
} |
|||
} catch (Throwable e) { |
|||
log.warn("Failed to obtain responses from queue.", e); |
|||
try { |
|||
Thread.sleep(pollInterval); |
|||
} catch (InterruptedException e2) { |
|||
log.trace("Failed to wait until the server has capacity to handle new responses", e2); |
|||
} |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
public void stop() { |
|||
stopped = true; |
|||
if (internalExecutor) { |
|||
executor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
public ListenableFuture<Response> post(String key, Request request) { |
|||
if (tickSize > maxPendingRequests) { |
|||
return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!")); |
|||
} |
|||
UUID requestId = UUID.randomUUID(); |
|||
List<Header> headers = new ArrayList<>(2); |
|||
headers.add(new RecordHeader(TbKafkaSettings.REQUEST_ID_HEADER, uuidToBytes(requestId))); |
|||
headers.add(new RecordHeader(TbKafkaSettings.RESPONSE_TOPIC_HEADER, stringToBytes(responseTemplate.getTopic()))); |
|||
SettableFuture<Response> future = SettableFuture.create(); |
|||
ResponseMetaData<Response> responseMetaData = new ResponseMetaData<>(tickTs + maxRequestTimeout, future); |
|||
pendingRequests.putIfAbsent(requestId, responseMetaData); |
|||
log.trace("[{}] Sending request, key [{}], expTime [{}]", requestId, key, responseMetaData.expTime); |
|||
requestTemplate.send(key, request, headers, (metadata, exception) -> { |
|||
if (exception != null) { |
|||
log.trace("[{}] Failed to post the request", requestId, exception); |
|||
} else { |
|||
log.trace("[{}] Posted the request: {}", requestId, metadata); |
|||
} |
|||
}); |
|||
return future; |
|||
} |
|||
|
|||
private static class ResponseMetaData<T> { |
|||
private final long expTime; |
|||
private final SettableFuture<T> future; |
|||
|
|||
ResponseMetaData(long ts, SettableFuture<T> future) { |
|||
this.expTime = ts; |
|||
this.future = future; |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,161 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.kafka.clients.consumer.ConsumerRecords; |
|||
import org.apache.kafka.common.errors.InterruptException; |
|||
import org.apache.kafka.common.header.Header; |
|||
import org.apache.kafka.common.header.internals.RecordHeader; |
|||
|
|||
import java.time.Duration; |
|||
import java.util.Collections; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeoutException; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
/** |
|||
* Created by ashvayka on 25.09.18. |
|||
*/ |
|||
@Slf4j |
|||
public class TbKafkaResponseTemplate<Request, Response> extends AbstractTbKafkaTemplate { |
|||
|
|||
private final TBKafkaConsumerTemplate<Request> requestTemplate; |
|||
private final TBKafkaProducerTemplate<Response> responseTemplate; |
|||
private final TbKafkaHandler<Request, Response> handler; |
|||
private final ConcurrentMap<UUID, String> pendingRequests; |
|||
private final ExecutorService loopExecutor; |
|||
private final ScheduledExecutorService timeoutExecutor; |
|||
private final ExecutorService callbackExecutor; |
|||
private final int maxPendingRequests; |
|||
private final long requestTimeout; |
|||
|
|||
private final long pollInterval; |
|||
private volatile boolean stopped = false; |
|||
private final AtomicInteger pendingRequestCount = new AtomicInteger(); |
|||
|
|||
@Builder |
|||
public TbKafkaResponseTemplate(TBKafkaConsumerTemplate<Request> requestTemplate, |
|||
TBKafkaProducerTemplate<Response> responseTemplate, |
|||
TbKafkaHandler<Request, Response> handler, |
|||
long pollInterval, |
|||
long requestTimeout, |
|||
int maxPendingRequests, |
|||
ExecutorService executor) { |
|||
this.requestTemplate = requestTemplate; |
|||
this.responseTemplate = responseTemplate; |
|||
this.handler = handler; |
|||
this.pendingRequests = new ConcurrentHashMap<>(); |
|||
this.maxPendingRequests = maxPendingRequests; |
|||
this.pollInterval = pollInterval; |
|||
this.requestTimeout = requestTimeout; |
|||
this.callbackExecutor = executor; |
|||
this.timeoutExecutor = Executors.newSingleThreadScheduledExecutor(); |
|||
this.loopExecutor = Executors.newSingleThreadExecutor(); |
|||
} |
|||
|
|||
public void init() { |
|||
this.responseTemplate.init(); |
|||
requestTemplate.subscribe(); |
|||
loopExecutor.submit(() -> { |
|||
while (!stopped) { |
|||
try { |
|||
while (pendingRequestCount.get() >= maxPendingRequests) { |
|||
try { |
|||
Thread.sleep(pollInterval); |
|||
} catch (InterruptedException e) { |
|||
log.trace("Failed to wait until the server has capacity to handle new requests", e); |
|||
} |
|||
} |
|||
ConsumerRecords<String, byte[]> requests = requestTemplate.poll(Duration.ofMillis(pollInterval)); |
|||
requests.forEach(request -> { |
|||
Header requestIdHeader = request.headers().lastHeader(TbKafkaSettings.REQUEST_ID_HEADER); |
|||
if (requestIdHeader == null) { |
|||
log.error("[{}] Missing requestId in header", request); |
|||
return; |
|||
} |
|||
UUID requestId = bytesToUuid(requestIdHeader.value()); |
|||
if (requestId == null) { |
|||
log.error("[{}] Missing requestId in header and body", request); |
|||
return; |
|||
} |
|||
Header responseTopicHeader = request.headers().lastHeader(TbKafkaSettings.RESPONSE_TOPIC_HEADER); |
|||
if (responseTopicHeader == null) { |
|||
log.error("[{}] Missing response topic in header", request); |
|||
return; |
|||
} |
|||
String responseTopic = bytesToString(responseTopicHeader.value()); |
|||
try { |
|||
pendingRequestCount.getAndIncrement(); |
|||
Request decodedRequest = requestTemplate.decode(request); |
|||
AsyncCallbackTemplate.withCallbackAndTimeout(handler.handle(decodedRequest), |
|||
response -> { |
|||
pendingRequestCount.decrementAndGet(); |
|||
reply(requestId, responseTopic, response); |
|||
}, |
|||
e -> { |
|||
pendingRequestCount.decrementAndGet(); |
|||
if (e.getCause() != null && e.getCause() instanceof TimeoutException) { |
|||
log.warn("[{}] Timedout to process the request: {}", requestId, request, e); |
|||
} else { |
|||
log.trace("[{}] Failed to process the request: {}", requestId, request, e); |
|||
} |
|||
}, |
|||
requestTimeout, |
|||
timeoutExecutor, |
|||
callbackExecutor); |
|||
} catch (Throwable e) { |
|||
pendingRequestCount.decrementAndGet(); |
|||
log.warn("[{}] Failed to process the request: {}", requestId, request, e); |
|||
} |
|||
}); |
|||
} catch (InterruptException ie) { |
|||
if (!stopped) { |
|||
log.warn("Fetching data from kafka was interrupted.", ie); |
|||
} |
|||
} catch (Throwable e) { |
|||
log.warn("Failed to obtain messages from queue.", e); |
|||
try { |
|||
Thread.sleep(pollInterval); |
|||
} catch (InterruptedException e2) { |
|||
log.trace("Failed to wait until the server has capacity to handle new requests", e2); |
|||
} |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
public void stop() { |
|||
stopped = true; |
|||
if (timeoutExecutor != null) { |
|||
timeoutExecutor.shutdownNow(); |
|||
} |
|||
if (loopExecutor != null) { |
|||
loopExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
private void reply(UUID requestId, String topic, Response response) { |
|||
responseTemplate.send(topic, requestId.toString(), response, Collections.singletonList(new RecordHeader(TbKafkaSettings.REQUEST_ID_HEADER, uuidToBytes(requestId))), null); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,79 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.provider; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.TbQueueConsumer; |
|||
import org.thingsboard.server.TbQueueCoreSettings; |
|||
import org.thingsboard.server.TbQueueProducer; |
|||
import org.thingsboard.server.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.memory.InMemoryTbQueueConsumer; |
|||
import org.thingsboard.server.memory.InMemoryTbQueueProducer; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
@ConditionalOnExpression("('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core') && '${queue.type:null}'=='in-memory'") |
|||
public class InMemoryTbCoreQueueProvider implements TbCoreQueueProvider { |
|||
|
|||
private final TbQueueCoreSettings coreSettings; |
|||
|
|||
public InMemoryTbCoreQueueProvider(TbQueueCoreSettings coreSettings) { |
|||
this.coreSettings = coreSettings; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportMsgProducer() { |
|||
InMemoryTbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); |
|||
return producer; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
InMemoryTbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); |
|||
return producer; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
InMemoryTbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); |
|||
return producer; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> getToCoreMsgConsumer() { |
|||
InMemoryTbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); |
|||
return consumer; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> getTransportApiRequestConsumer() { |
|||
InMemoryTbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); |
|||
return consumer; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiResponseProducer() { |
|||
InMemoryTbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); |
|||
return producer; |
|||
} |
|||
} |
|||
@ -0,0 +1,79 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.provider; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.TbQueueConsumer; |
|||
import org.thingsboard.server.TbQueueProducer; |
|||
import org.thingsboard.server.TbQueueRequestTemplate; |
|||
import org.thingsboard.server.TbQueueTransportApiSettings; |
|||
import org.thingsboard.server.common.DefaultTbQueueRequestTemplate; |
|||
import org.thingsboard.server.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.memory.InMemoryTbQueueConsumer; |
|||
import org.thingsboard.server.memory.InMemoryTbQueueProducer; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport') && '${queue.type:null}'=='in-memory'") |
|||
@Slf4j |
|||
public class InMemoryTransportQueueProvider implements TransportQueueProvider { |
|||
|
|||
private final TbQueueTransportApiSettings transportApiSettings; |
|||
|
|||
public InMemoryTransportQueueProvider(TbQueueTransportApiSettings transportApiSettings) { |
|||
this.transportApiSettings = transportApiSettings; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiRequestTemplate() { |
|||
InMemoryTbQueueProducer<TbProtoQueueMsg<TransportApiRequestMsg>> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); |
|||
|
|||
InMemoryTbQueueConsumer<TbProtoQueueMsg<TransportApiResponseMsg>> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic()); |
|||
|
|||
DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder |
|||
<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder(); |
|||
templateBuilder.requestTemplate(producer); |
|||
templateBuilder.responseTemplate(consumer); |
|||
templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); |
|||
templateBuilder.maxRequestTimeout(transportApiSettings.getMaxRequestsTimeout()); |
|||
templateBuilder.pollInterval(transportApiSettings.getResponsePollInterval()); |
|||
return templateBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
InMemoryTbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); |
|||
return producer; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
InMemoryTbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); |
|||
return producer; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsConsumer() { |
|||
InMemoryTbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic()); |
|||
return consumer; |
|||
} |
|||
} |
|||
@ -1,42 +1,102 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.TbQueueConsumer; |
|||
import org.thingsboard.server.TbQueueCoreSettings; |
|||
import org.thingsboard.server.TbQueueProducer; |
|||
import org.thingsboard.server.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; |
|||
import org.thingsboard.server.kafka.TBKafkaProducerTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.kafka.TbNodeIdProvider; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core') && ${queue.type:null}'=='kafka'") |
|||
public class KafkaTbCoreQueueProvider implements TbCoreQueueProvider{ |
|||
@ConditionalOnExpression("('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core') && '${queue.type:null}'=='kafka'") |
|||
public class KafkaTbCoreQueueProvider implements TbCoreQueueProvider { |
|||
|
|||
private final TbKafkaSettings kafkaSettings; |
|||
private final TbNodeIdProvider nodeIdProvider; |
|||
private final TbQueueCoreSettings coreSettings; |
|||
|
|||
public KafkaTbCoreQueueProvider(TbKafkaSettings kafkaSettings, TbNodeIdProvider nodeIdProvider, TbQueueCoreSettings coreSettings) { |
|||
this.kafkaSettings = kafkaSettings; |
|||
this.nodeIdProvider = nodeIdProvider; |
|||
this.coreSettings = coreSettings; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> getTransportMsgProducer() { |
|||
return null; |
|||
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToTransportMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return null; |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToRuleEngineMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return null; |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToCoreMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getToCoreMsgConsumer() { |
|||
return null; |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> getToCoreMsgConsumer() { |
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToCoreMsg>> responseBuilder = TBKafkaConsumerTemplate.builder(); |
|||
responseBuilder.settings(kafkaSettings); |
|||
responseBuilder.topic(coreSettings.getTopic()); |
|||
responseBuilder.clientId("consumer-transport-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.groupId("rule-engine-node-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.autoCommit(true); |
|||
//TODO: 2.5
|
|||
// responseBuilder.autoCommitIntervalMs(autoCommitInterval);
|
|||
// responseBuilder.decoder(new TransportApiResponseDecoder());
|
|||
return responseBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>> getTransportApiRequestConsumer() { |
|||
public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> getTransportApiRequestConsumer() { |
|||
return null; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> getTransportApiResponseProducer() { |
|||
return null; |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiResponseProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<TransportApiResponseMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
} |
|||
|
|||
Loading…
Reference in new issue