Browse Source
* rest api call node added redis queue support * rest api node refactoringpull/2138/head
Dima Landiak
7 years ago
committed by
Andrew Shvayka
Andrew Shvayka
10 changed files with 354 additions and 110 deletions
@ -0,0 +1,153 @@ |
|||
/** |
|||
* Copyright © 2016-2019 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.rule.engine.rest; |
|||
|
|||
import io.netty.channel.EventLoopGroup; |
|||
import io.netty.channel.nio.NioEventLoopGroup; |
|||
import io.netty.handler.ssl.SslContextBuilder; |
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.http.HttpEntity; |
|||
import org.springframework.http.HttpHeaders; |
|||
import org.springframework.http.HttpMethod; |
|||
import org.springframework.http.ResponseEntity; |
|||
import org.springframework.http.client.Netty4ClientHttpRequestFactory; |
|||
import org.springframework.util.concurrent.ListenableFuture; |
|||
import org.springframework.util.concurrent.ListenableFutureCallback; |
|||
import org.springframework.web.client.AsyncRestTemplate; |
|||
import org.springframework.web.client.HttpClientErrorException; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.TbNodeException; |
|||
import org.thingsboard.rule.engine.api.TbRelationTypes; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
|
|||
import javax.net.ssl.SSLException; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Data |
|||
@Slf4j |
|||
class TbHttpClient { |
|||
|
|||
private static final String STATUS = "status"; |
|||
private static final String STATUS_CODE = "statusCode"; |
|||
private static final String STATUS_REASON = "statusReason"; |
|||
private static final String ERROR = "error"; |
|||
private static final String ERROR_BODY = "error_body"; |
|||
|
|||
private final TbRestApiCallNodeConfiguration config; |
|||
private final boolean useRedisQueueForMsgPersistence; |
|||
|
|||
private EventLoopGroup eventLoopGroup; |
|||
private AsyncRestTemplate httpClient; |
|||
|
|||
TbHttpClient(TbRestApiCallNodeConfiguration config) throws TbNodeException { |
|||
try { |
|||
this.config = config; |
|||
this.useRedisQueueForMsgPersistence = config.isUseRedisQueueForMsgPersistence(); |
|||
if (config.isUseSimpleClientHttpFactory()) { |
|||
httpClient = new AsyncRestTemplate(); |
|||
} else { |
|||
this.eventLoopGroup = new NioEventLoopGroup(); |
|||
Netty4ClientHttpRequestFactory nettyFactory = new Netty4ClientHttpRequestFactory(this.eventLoopGroup); |
|||
nettyFactory.setSslContext(SslContextBuilder.forClient().build()); |
|||
httpClient = new AsyncRestTemplate(nettyFactory); |
|||
} |
|||
} catch (SSLException e) { |
|||
throw new TbNodeException(e); |
|||
} |
|||
} |
|||
|
|||
void destroy() { |
|||
if (this.eventLoopGroup != null) { |
|||
this.eventLoopGroup.shutdownGracefully(0, 5, TimeUnit.SECONDS); |
|||
} |
|||
} |
|||
|
|||
void processMessage(TbContext ctx, TbMsg msg, TbRedisQueueProcessor queueProcessor) { |
|||
String endpointUrl = TbNodeUtils.processPattern(config.getRestEndpointUrlPattern(), msg.getMetaData()); |
|||
HttpHeaders headers = prepareHeaders(msg.getMetaData()); |
|||
HttpMethod method = HttpMethod.valueOf(config.getRequestMethod()); |
|||
HttpEntity<String> entity = new HttpEntity<>(msg.getData(), headers); |
|||
|
|||
ListenableFuture<ResponseEntity<String>> future = httpClient.exchange( |
|||
endpointUrl, method, entity, String.class); |
|||
future.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() { |
|||
@Override |
|||
public void onFailure(Throwable throwable) { |
|||
if (useRedisQueueForMsgPersistence) { |
|||
queueProcessor.pushOnFailure(msg); |
|||
} |
|||
TbMsg next = processException(ctx, msg, throwable); |
|||
ctx.tellFailure(next, throwable); |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess(ResponseEntity<String> responseEntity) { |
|||
if (responseEntity.getStatusCode().is2xxSuccessful()) { |
|||
if (useRedisQueueForMsgPersistence) { |
|||
queueProcessor.resetCounter(); |
|||
} |
|||
TbMsg next = processResponse(ctx, msg, responseEntity); |
|||
ctx.tellNext(next, TbRelationTypes.SUCCESS); |
|||
} else { |
|||
if (useRedisQueueForMsgPersistence) { |
|||
queueProcessor.pushOnFailure(msg); |
|||
} |
|||
TbMsg next = processFailureResponse(ctx, msg, responseEntity); |
|||
ctx.tellNext(next, TbRelationTypes.FAILURE); |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private TbMsg processResponse(TbContext ctx, TbMsg origMsg, ResponseEntity<String> response) { |
|||
TbMsgMetaData metaData = origMsg.getMetaData(); |
|||
metaData.putValue(STATUS, response.getStatusCode().name()); |
|||
metaData.putValue(STATUS_CODE, response.getStatusCode().value() + ""); |
|||
metaData.putValue(STATUS_REASON, response.getStatusCode().getReasonPhrase()); |
|||
response.getHeaders().toSingleValueMap().forEach(metaData::putValue); |
|||
return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, response.getBody()); |
|||
} |
|||
|
|||
private TbMsg processFailureResponse(TbContext ctx, TbMsg origMsg, ResponseEntity<String> response) { |
|||
TbMsgMetaData metaData = origMsg.getMetaData(); |
|||
metaData.putValue(STATUS, response.getStatusCode().name()); |
|||
metaData.putValue(STATUS_CODE, response.getStatusCode().value() + ""); |
|||
metaData.putValue(STATUS_REASON, response.getStatusCode().getReasonPhrase()); |
|||
metaData.putValue(ERROR_BODY, response.getBody()); |
|||
return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData()); |
|||
} |
|||
|
|||
private TbMsg processException(TbContext ctx, TbMsg origMsg, Throwable e) { |
|||
TbMsgMetaData metaData = origMsg.getMetaData(); |
|||
metaData.putValue(ERROR, e.getClass() + ": " + e.getMessage()); |
|||
if (e instanceof HttpClientErrorException) { |
|||
HttpClientErrorException httpClientErrorException = (HttpClientErrorException) e; |
|||
metaData.putValue(STATUS, httpClientErrorException.getStatusText()); |
|||
metaData.putValue(STATUS_CODE, httpClientErrorException.getRawStatusCode() + ""); |
|||
metaData.putValue(ERROR_BODY, httpClientErrorException.getResponseBodyAsString()); |
|||
} |
|||
return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData()); |
|||
} |
|||
|
|||
private HttpHeaders prepareHeaders(TbMsgMetaData metaData) { |
|||
HttpHeaders headers = new HttpHeaders(); |
|||
config.getHeaders().forEach((k, v) -> headers.add(TbNodeUtils.processPattern(k, metaData), TbNodeUtils.processPattern(v, metaData))); |
|||
return headers; |
|||
} |
|||
} |
|||
@ -0,0 +1,125 @@ |
|||
/** |
|||
* Copyright © 2016-2019 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.rule.engine.rest; |
|||
|
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.data.redis.core.ListOperations; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import java.util.List; |
|||
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.AtomicInteger; |
|||
|
|||
@Data |
|||
@Slf4j |
|||
class TbRedisQueueProcessor { |
|||
|
|||
private static final int MAX_QUEUE_SIZE = Integer.MAX_VALUE; |
|||
|
|||
private final TbContext ctx; |
|||
private final TbHttpClient httpClient; |
|||
private final ExecutorService executor; |
|||
private final ListOperations<String, Object> listOperations; |
|||
private final String redisKey; |
|||
private final boolean trimQueue; |
|||
private final int maxQueueSize; |
|||
|
|||
private AtomicInteger failuresCounter; |
|||
private Future future; |
|||
|
|||
TbRedisQueueProcessor(TbContext ctx, TbHttpClient httpClient, boolean trimQueue, int maxQueueSize) { |
|||
this.ctx = ctx; |
|||
this.httpClient = httpClient; |
|||
this.executor = Executors.newSingleThreadExecutor(); |
|||
this.listOperations = ctx.getRedisTemplate().opsForList(); |
|||
this.redisKey = constructRedisKey(); |
|||
this.trimQueue = trimQueue; |
|||
this.maxQueueSize = maxQueueSize; |
|||
init(); |
|||
} |
|||
|
|||
private void init() { |
|||
failuresCounter = new AtomicInteger(0); |
|||
future = executor.submit(() -> { |
|||
while (true) { |
|||
if (failuresCounter.get() != 0 && failuresCounter.get() % 50 == 0) { |
|||
sleep("Target HTTP server is down...", 3); |
|||
} |
|||
if (listOperations.size(redisKey) > 0) { |
|||
List<Object> list = listOperations.range(redisKey, -10, -1); |
|||
list.forEach(obj -> { |
|||
TbMsg msg = TbMsg.fromBytes((byte[]) obj); |
|||
log.debug("Trying to send the message: {}", msg); |
|||
listOperations.remove(redisKey, -1, obj); |
|||
httpClient.processMessage(ctx, msg, this); |
|||
}); |
|||
} else { |
|||
sleep("Queue is empty, waiting for tasks!", 1); |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
void destroy() { |
|||
if (future != null) { |
|||
future.cancel(true); |
|||
} |
|||
if (executor != null) { |
|||
executor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
void push(TbMsg msg) { |
|||
listOperations.leftPush(redisKey, TbMsg.toByteArray(msg)); |
|||
if (trimQueue) { |
|||
listOperations.trim(redisKey, 0, validateMaxQueueSize()); |
|||
} |
|||
} |
|||
|
|||
void pushOnFailure(TbMsg msg) { |
|||
listOperations.rightPush(redisKey, TbMsg.toByteArray(msg)); |
|||
failuresCounter.incrementAndGet(); |
|||
} |
|||
|
|||
void resetCounter() { |
|||
failuresCounter.set(0); |
|||
} |
|||
|
|||
private String constructRedisKey() { |
|||
return ctx.getServerAddress() + ctx.getSelfId(); |
|||
} |
|||
|
|||
private int validateMaxQueueSize() { |
|||
if (maxQueueSize != 0) { |
|||
return maxQueueSize; |
|||
} |
|||
return MAX_QUEUE_SIZE; |
|||
} |
|||
|
|||
private void sleep(String logMessage, int sleepSeconds) { |
|||
try { |
|||
log.debug(logMessage); |
|||
TimeUnit.SECONDS.sleep(sleepSeconds); |
|||
} catch (InterruptedException e) { |
|||
throw new IllegalStateException("Thread interrupted!", e); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue