fixed the issues

revert-70aa11f8
lashanfaliq95 7 years ago
parent 65cfca7024
commit 7a1c17081b

@ -294,8 +294,6 @@ public interface DeviceEventManagementService {
@ApiParam(name = "limit", value = "limit of the records that needs to be picked up", required = false) @ApiParam(name = "limit", value = "limit of the records that needs to be picked up", required = false)
@QueryParam("limit") int limit); @QueryParam("limit") int limit);
@GET @GET
@Path("/{type}") @Path("/{type}")
@ApiOperation( @ApiOperation(

@ -76,8 +76,47 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
private static final String MQTT_CONTENT_VALIDATOR = "default"; private static final String MQTT_CONTENT_VALIDATOR = "default";
private static final String TIMESTAMP_FIELD_NAME = "_timestamp"; private static final String TIMESTAMP_FIELD_NAME = "_timestamp";
private static AnalyticsDataAPI getAnalyticsDataAPI() {
PrivilegedCarbonContext ctx = PrivilegedCarbonContext.getThreadLocalCarbonContext();
AnalyticsDataAPI analyticsDataAPI =
(AnalyticsDataAPI) ctx.getOSGiService(AnalyticsDataAPI.class, null);
if (analyticsDataAPI == null) {
String msg = "Analytics api service has not initialized.";
log.error(msg);
throw new IllegalStateException(msg);
}
return analyticsDataAPI;
}
private static EventRecords getAllEventsForDevice(String tableName, String query, List<SortByField> sortByFields
, int offset, int limit) throws AnalyticsException {
int tenantId = CarbonContext.getThreadLocalCarbonContext().getTenantId();
AnalyticsDataAPI analyticsDataAPI = getAnalyticsDataAPI();
EventRecords eventRecords = new EventRecords();
int eventCount = analyticsDataAPI.searchCount(tenantId, tableName, query);
if (eventCount == 0) {
eventRecords.setCount(0);
}
List<SearchResultEntry> resultEntries = analyticsDataAPI.search(tenantId, tableName, query, offset, limit,
sortByFields);
List<String> recordIds = getRecordIds(resultEntries);
AnalyticsDataResponse response = analyticsDataAPI.get(tenantId, tableName, 1, null, recordIds);
eventRecords.setCount(eventCount);
eventRecords.setList(AnalyticsDataAPIUtil.listRecords(analyticsDataAPI, response));
return eventRecords;
}
private static List<String> getRecordIds(List<SearchResultEntry> searchResults) {
List<String> ids = new ArrayList<>();
for (SearchResultEntry searchResult : searchResults) {
ids.add(searchResult.getId());
}
return ids;
}
/** /**
* Retrieves the stream definition from das for the given device type. * Retrieves the stream definition from das for the given device type.
*
* @return dynamic event attribute list * @return dynamic event attribute list
*/ */
@GET @GET
@ -229,9 +268,11 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
} }
eventStreamAdminServiceStub.removeEventStreamDefinition(streamName, Constants.DEFAULT_STREAM_VERSION); eventStreamAdminServiceStub.removeEventStreamDefinition(streamName, Constants.DEFAULT_STREAM_VERSION);
EventReceiverAdminServiceCallbackHandler eventReceiverAdminServiceCallbackHandler = EventReceiverAdminServiceCallbackHandler eventReceiverAdminServiceCallbackHandler =
new EventReceiverAdminServiceCallbackHandler() {}; new EventReceiverAdminServiceCallbackHandler() {
};
EventPublisherAdminServiceCallbackHandler eventPublisherAdminServiceCallbackHandler = EventPublisherAdminServiceCallbackHandler eventPublisherAdminServiceCallbackHandler =
new EventPublisherAdminServiceCallbackHandler() {}; new EventPublisherAdminServiceCallbackHandler() {
};
String eventReceiverName = getReceiverName(deviceType, tenantDomain, TransportType.MQTT); String eventReceiverName = getReceiverName(deviceType, tenantDomain, TransportType.MQTT);
@ -365,8 +406,8 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
sortByFields.add(sortByField); sortByFields.add(sortByField);
if (limit == 0) { if (limit == 0) {
EventRecords eventRecords = getAllEventsForDevice(sensorTableName, query, sortByFields, 0, 1); EventRecords eventRecords = getAllEventsForDevice(sensorTableName, query, sortByFields, 0, 1);
return Response.status(Response.Status.OK.getStatusCode()).entity(eventRecords).build();} return Response.status(Response.Status.OK.getStatusCode()).entity(eventRecords).build();
else{ } else {
EventRecords eventRecords = getAllEventsForDevice(sensorTableName, query, sortByFields, 0, limit); EventRecords eventRecords = getAllEventsForDevice(sensorTableName, query, sortByFields, 0, limit);
return Response.status(Response.Status.OK.getStatusCode()).entity(eventRecords).build(); return Response.status(Response.Status.OK.getStatusCode()).entity(eventRecords).build();
} }
@ -384,7 +425,6 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
} }
} }
private void publishEventReceivers(String streamNameWithVersion, TransportType transportType private void publishEventReceivers(String streamNameWithVersion, TransportType transportType
, String requestedTenantDomain, String deviceType) , String requestedTenantDomain, String deviceType)
throws RemoteException, UserStoreException, JWTClientException { throws RemoteException, UserStoreException, JWTClientException {
@ -399,7 +439,8 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
.getActiveEventReceiverConfiguration(eventRecieverNameTobeRemoved); .getActiveEventReceiverConfiguration(eventRecieverNameTobeRemoved);
if (eventReceiverConfigurationDto != null) { if (eventReceiverConfigurationDto != null) {
EventReceiverAdminServiceCallbackHandler eventReceiverAdminServiceCallbackHandler = EventReceiverAdminServiceCallbackHandler eventReceiverAdminServiceCallbackHandler =
new EventReceiverAdminServiceCallbackHandler() {}; new EventReceiverAdminServiceCallbackHandler() {
};
receiverAdminServiceStub.startundeployActiveEventReceiverConfiguration(eventRecieverNameTobeRemoved receiverAdminServiceStub.startundeployActiveEventReceiverConfiguration(eventRecieverNameTobeRemoved
, eventReceiverAdminServiceCallbackHandler); , eventReceiverAdminServiceCallbackHandler);
} }
@ -545,44 +586,6 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
return deviceType.replace(" ", "_").trim() + "-" + tenantDomain + "-" + transportType.toString() + "-receiver"; return deviceType.replace(" ", "_").trim() + "-" + tenantDomain + "-" + transportType.toString() + "-receiver";
} }
private static AnalyticsDataAPI getAnalyticsDataAPI() {
PrivilegedCarbonContext ctx = PrivilegedCarbonContext.getThreadLocalCarbonContext();
AnalyticsDataAPI analyticsDataAPI =
(AnalyticsDataAPI) ctx.getOSGiService(AnalyticsDataAPI.class, null);
if (analyticsDataAPI == null) {
String msg = "Analytics api service has not initialized.";
log.error(msg);
throw new IllegalStateException(msg);
}
return analyticsDataAPI;
}
private static EventRecords getAllEventsForDevice(String tableName, String query, List<SortByField> sortByFields
, int offset, int limit) throws AnalyticsException {
int tenantId = CarbonContext.getThreadLocalCarbonContext().getTenantId();
AnalyticsDataAPI analyticsDataAPI = getAnalyticsDataAPI();
EventRecords eventRecords = new EventRecords();
int eventCount = analyticsDataAPI.searchCount(tenantId, tableName, query);
if (eventCount == 0) {
eventRecords.setCount(0);
}
List<SearchResultEntry> resultEntries = analyticsDataAPI.search(tenantId, tableName, query, offset, limit,
sortByFields);
List<String> recordIds = getRecordIds(resultEntries);
AnalyticsDataResponse response = analyticsDataAPI.get(tenantId, tableName, 1, null, recordIds);
eventRecords.setCount(eventCount);
eventRecords.setList(AnalyticsDataAPIUtil.listRecords(analyticsDataAPI, response));
return eventRecords;
}
private static List<String> getRecordIds(List<SearchResultEntry> searchResults) {
List<String> ids = new ArrayList<>();
for (SearchResultEntry searchResult : searchResults) {
ids.add(searchResult.getId());
}
return ids;
}
private void cleanup(Stub stub) { private void cleanup(Stub stub) {
if (stub != null) { if (stub != null) {
try { try {

Loading…
Cancel
Save