Skip to content

G
gic-platform-message
Commit eef4e911 by 陶光胜
Options

init

parent cde6c09e
Showing with 296 additions and 78 deletions
src/main/java/com/gic/message/service/impl/KafkaMessageServiceImpl.java
......@@ -5,17 +5,17 @@ import com.gic.binlog.base.entity.GicRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.MessageListener;
public class KafkaMessageServiceImpl implements MessageListener<String, GicRecord> {
private static final Logger log = LogManager.getLogger(KafkaMessageServiceImpl.class);
@Autowired
private MessageHandlerInit messageHandlerInit;
@Override
public void onMessage(ConsumerRecord<String, GicRecord> consumerRecord) {
log.info("接收kafka消息，内容为111："+consumerRecord.value());
}
public static void main(String[] args){
System.out.println("110101".substring(0, "110101".length()-2));
this.messageHandlerInit.getHandler().get(consumerRecord.value().getTableName()).handler(consumerRecord);
}
}
src/main/java/com/gic/message/service/impl/MessageHandlerInit.java 0 → 100644
package com.gic.message.service.impl;
import com.gic.message.service.MessageHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
@Component
public class MessageHandlerInit {
@Autowired
private StoreMessageHandler storeMessageHandler;
@Autowired
private StoreInfoMessageHandler storeInfoMessageHandler;
@Autowired
private StoreCustomMessageHandler storeCustomMessageHandler;
private Map<String, MessageHandler> handler = new HashMap<>();
@PostConstruct
public void init(){
handler.put("tab_store", this.storeMessageHandler);
handler.put("tab_store_info", this.storeInfoMessageHandler);
handler.put("tab_store_extend", this.storeCustomMessageHandler);
}
public Map<String, MessageHandler> getHandler() {
return handler;
}
public void setHandler(Map<String, MessageHandler> handler) {
this.handler = handler;
}
}
src/main/java/com/gic/message/service/impl/StoreCustomMessageHandler.java 0 → 100644
package com.gic.message.service.impl;
import com.gic.api.base.commons.Page;
import com.gic.api.base.commons.ServiceResponse;
import com.gic.auth.constant.AuthModeEnum;
import com.gic.auth.dto.UnionEnterpriseResourceDTO;
import com.gic.auth.service.UnionEnterpriseApiService;
import com.gic.binlog.base.entity.GicField;
import com.gic.binlog.base.entity.GicRecord;
import com.gic.binlog.base.entity.enums.GicRecordType;
import com.gic.message.service.MessageHandler;
import com.gic.message.utils.ListToMapUtil;
import com.gic.redis.data.util.RedisUtil;
import com.gic.store.constant.StoreOwnTypeEnum;
import com.gic.store.dto.StoreDTO;
import com.gic.store.dto.StoreSearchDTO;
import com.gic.store.dto.StoreWidgetDTO;
import com.gic.store.service.StoreApiService;
import com.gic.store.service.StoreAuthorizationApiService;
import com.gic.store.service.StoreWidgetApiService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
@Component
public class StoreCustomMessageHandler implements MessageHandler {
@Autowired
private StoreApiService storeApiService;
@Autowired
private StoreIndexRefreshHandler storeIndexRefreshHandler;
@Autowired
private StoreAuthorizationApiService storeAuthorizationApiService;
@Autowired
private UnionEnterpriseApiService unionEnterpriseApiService;
@Autowired
private StoreWidgetApiService storeWidgetApiService;
@Override
public void handler(ConsumerRecord<String, GicRecord> consumerRecord) {
GicRecord value = consumerRecord.value();
Map<String, GicField> fieldMap = ListToMapUtil.listToMap(consumerRecord);
Integer enterpriseId = 0, storeInfoId = 0, storeFieldId = 0;
if(GicRecordType.INSERT.value() == value.getRecordType().value() ||
GicRecordType.DELETE.value() == value.getRecordType().value()){
enterpriseId = Integer.valueOf(fieldMap.get("enterprise_id").getValue());
storeFieldId = Integer.valueOf(fieldMap.get("store_field_id").getValue());
storeInfoId = Integer.valueOf(fieldMap.get("store_info_id").getValue());
this.dealCustomChange(enterpriseId, storeFieldId, storeInfoId, fieldMap);
}
if(GicRecordType.UPDATE.value() == value.getRecordType().value()){
enterpriseId = Integer.valueOf(fieldMap.get("new_enterprise_id").getValue());
storeFieldId = Integer.valueOf(fieldMap.get("new_store_field_id").getValue());
storeInfoId = Integer.valueOf(fieldMap.get("new_store_info_id").getValue());
if(!fieldMap.get("old_value").getValue().equals(fieldMap.get("new_value").getValue())){
this.dealCustomChange(enterpriseId, storeFieldId, storeInfoId, fieldMap);
}
}
}
private void dealCustomChange(Integer enterpriseId, Integer storeFieldId, Integer storeInfoId, Map<String, GicField> fieldMap){
String key = "enterprise:store:storeField:" + enterpriseId + ":" + storeFieldId;
Object existObj = RedisUtil.getCache(key);
if(existObj != null){
ServiceResponse<List<StoreDTO>> listStoreResponse = this.storeApiService.listStoreByStoreInfoId(storeInfoId);
List<StoreDTO> listStore = listStoreResponse.getResult();
for(StoreDTO storeDTO : listStore){
if(StoreOwnTypeEnum.OWNER.getCode() == storeDTO.getOwnType()){
this.storeIndexRefreshHandler.refreshStoreIndex(storeDTO.getEnterpriseId(), storeDTO.getStoreId());
}
}
this.dealStoreInfoChange(enterpriseId, storeInfoId, fieldMap);
ServiceResponse<List<StoreDTO>> listStoreResponse1 = this.storeApiService.listStoreByStoreInfoId(storeInfoId);
List<StoreDTO> listStore1 = listStoreResponse1.getResult();
for(StoreDTO storeDTO : listStore1){
this.storeIndexRefreshHandler.refreshStoreIndex(storeDTO.getEnterpriseId(), storeDTO.getStoreId());
}
}
}
private void dealStoreInfoChange(Integer enterpriseId, Integer storeInfoId, Map<String, GicField> fieldMap){
ServiceResponse<List<UnionEnterpriseResourceDTO>> resourceResponse = this.unionEnterpriseApiService.listStoreResourceByEnterpriseId(enterpriseId);
if(CollectionUtils.isNotEmpty(resourceResponse.getResult())){
for(UnionEnterpriseResourceDTO dto : resourceResponse.getResult()){
ServiceResponse<StoreWidgetDTO> storeWidget = this.storeWidgetApiService.getStoreWidget(dto.getResource().intValue());
if(!storeWidget.isSuccess()){
continue;
}
StoreWidgetDTO storeWidgetDTO = storeWidget.getResult();
if(storeWidgetDTO != null){
if(AuthModeEnum.YES.getCode() == storeWidgetDTO.getAuthMode()){
StoreSearchDTO storeSearchDTO = new StoreSearchDTO();
storeSearchDTO.setStoreInfoIds(String.valueOf(storeInfoId));
storeSearchDTO.setSearchJson(storeWidgetDTO.getSearchParam());
storeSearchDTO.setEnterpriseId(enterpriseId);
ServiceResponse<Page<StoreDTO>> response = this.storeApiService.listStore(storeSearchDTO, 1, 1);
if(!response.isSuccess()){
continue;
}
if(CollectionUtils.isNotEmpty(response.getResult().getResult())){
StoreDTO storeDTO = new StoreDTO();
storeDTO.setStoreInfoId(storeInfoId);
storeDTO.setEnterpriseId(enterpriseId);
storeDTO.setFromEnterpriseId(dto.getUnionEnterpriseId());
this.storeAuthorizationApiService.saveStore(storeDTO);
}else {
this.storeAuthorizationApiService.cancelStore(storeInfoId, dto.getUnionEnterpriseId());
}
}
}
}
}
}
}
src/main/java/com/gic/message/service/impl/StoreInfoMessageHandler.java
package com.gic.message.service.impl;
import com.gic.api.base.commons.Page;
import com.gic.api.base.commons.ServiceResponse;
import com.gic.auth.constant.AuthModeEnum;
import com.gic.auth.dto.UnionEnterpriseResourceDTO;
import com.gic.auth.service.UnionEnterpriseApiService;
import com.gic.binlog.base.entity.GicField;
import com.gic.binlog.base.entity.GicRecord;
import com.gic.binlog.base.entity.enums.GicRecordType;
......@@ -8,7 +12,13 @@ import com.gic.message.service.MessageHandler;
import com.gic.message.utils.ListToMapUtil;
import com.gic.store.constant.StoreOwnTypeEnum;
import com.gic.store.dto.StoreDTO;
import com.gic.store.dto.StoreSearchDTO;
import com.gic.store.dto.StoreWidgetDTO;
import com.gic.store.service.StoreApiService;
import com.gic.store.service.StoreAuthorizationApiService;
import com.gic.store.service.StoreRegionApiService;
import com.gic.store.service.StoreWidgetApiService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
......@@ -22,6 +32,14 @@ public class StoreInfoMessageHandler implements MessageHandler {
private StoreIndexRefreshHandler storeIndexRefreshHandler;
@Autowired
private StoreApiService storeApiService;
@Autowired
private StoreWidgetApiService storeWidgetApiService;
@Autowired
private UnionEnterpriseApiService unionEnterpriseApiService;
@Autowired
private StoreAuthorizationApiService storeAuthorizationApiService;
@Autowired
private StoreRegionApiService storeRegionApiService;
@Override
public void handler(ConsumerRecord<String, GicRecord> consumerRecord) {
......@@ -29,43 +47,81 @@ public class StoreInfoMessageHandler implements MessageHandler {
Map<String, GicField> fieldMap = ListToMapUtil.listToMap(consumerRecord);
Integer enterpriseId = 0, storeInfoId = 0, storeId = 0;
if(GicRecordType.INSERT.value() == value.getRecordType().value()){
storeInfoId = Integer.valueOf(fieldMap.get("store_info_id").getValue());
}
if(GicRecordType.UPDATE.value() == value.getRecordType().value()){
enterpriseId = Integer.valueOf(fieldMap.get("new_enterprise_id").getValue());
storeInfoId = Integer.valueOf(fieldMap.get("new_store_info_id").getValue());
ServiceResponse<List<StoreDTO>> listStoreResponse = this.storeApiService.listStoreByStoreInfoId(storeInfoId);
List<StoreDTO> listStore = listStoreResponse.getResult();
for(StoreDTO storeDTO : listStore){
enterpriseId = storeDTO.getEnterpriseId();
if(storeDTO.getOwnType() == StoreOwnTypeEnum.OWNER.getCode()){
}else {
this.storeIndexRefreshHandler.refreshStoreIndex(enterpriseId, storeDTO.getStoreId());
if(Integer.valueOf(fieldMap.get("old_status").getValue()) == 1 && Integer.valueOf(fieldMap.get("new_status").getValue()) == 0){
this.storeApiService.deleteStoreFromES(storeDTO.getEnterpriseId(), storeDTO.getStoreId());
}else {
this.storeIndexRefreshHandler.refreshStoreIndex(storeDTO.getEnterpriseId(), storeDTO.getStoreId());
}
}
}
}
if(GicRecordType.UPDATE.value() == value.getRecordType().value()){
int isAdd = 0;
if(Integer.valueOf(fieldMap.get("old_status").getValue()) == 0 && Integer.valueOf(fieldMap.get("new_status").getValue()) == 1){
isAdd = 1;
}
if(Integer.valueOf(fieldMap.get("old_status").getValue()) == 1 && Integer.valueOf(fieldMap.get("new_status").getValue()) == 0){
isAdd = 2;
}
enterpriseId = Integer.valueOf(fieldMap.get("new_enterprise_id").getValue());
storeId = Integer.valueOf(fieldMap.get("new_store_id").getValue());
if(isAdd != 0){
if(isAdd == 1){
this.storeIndexRefreshHandler.refreshStoreIndex(enterpriseId, storeId);
}else {
this.storeApiService.deleteStoreFromES(enterpriseId, storeId);
}
}else {
this.storeIndexRefreshHandler.refreshStoreIndex(enterpriseId, storeId);
this.dealStoreInfoChange(enterpriseId, storeInfoId, fieldMap);
ServiceResponse<List<StoreDTO>> listStoreResponse1 = this.storeApiService.listStoreByStoreInfoId(storeInfoId);
List<StoreDTO> listStore1 = listStoreResponse1.getResult();
for(StoreDTO storeDTO : listStore1){
this.storeIndexRefreshHandler.refreshStoreIndex(storeDTO.getEnterpriseId(), storeDTO.getStoreId());
}
}
if(GicRecordType.DELETE.value() == value.getRecordType().value()){
enterpriseId = Integer.valueOf(fieldMap.get("enterprise_id").getValue());
storeId = Integer.valueOf(fieldMap.get("store_id").getValue());
this.storeApiService.deleteStoreFromES(enterpriseId, storeId);
storeInfoId = Integer.valueOf(fieldMap.get("store_info_id").getValue());
ServiceResponse<List<StoreDTO>> listStoreResponse = this.storeApiService.listStoreByStoreInfoId(storeInfoId);
List<StoreDTO> listStore = listStoreResponse.getResult();
for(StoreDTO storeDTO : listStore){
this.storeApiService.deleteStoreFromES(storeDTO.getEnterpriseId(), storeDTO.getStoreId());
}
}
}
private void dealStoreInfoChange(Integer enterpriseId, Integer storeInfoId, Map<String, GicField> fieldMap){
ServiceResponse<List<UnionEnterpriseResourceDTO>> resourceResponse = this.unionEnterpriseApiService.listStoreResourceByEnterpriseId(enterpriseId);
if(CollectionUtils.isNotEmpty(resourceResponse.getResult())){
for(UnionEnterpriseResourceDTO dto : resourceResponse.getResult()){
ServiceResponse<StoreWidgetDTO> storeWidget = this.storeWidgetApiService.getStoreWidget(dto.getResource().intValue());
if(!storeWidget.isSuccess()){
continue;
}
StoreWidgetDTO storeWidgetDTO = storeWidget.getResult();
if(storeWidgetDTO != null){
if(AuthModeEnum.YES.getCode() == storeWidgetDTO.getAuthMode()){
StoreSearchDTO storeSearchDTO = new StoreSearchDTO();
storeSearchDTO.setStoreInfoIds(String.valueOf(storeInfoId));
storeSearchDTO.setSearchJson(storeWidgetDTO.getSearchParam());
storeSearchDTO.setEnterpriseId(enterpriseId);
ServiceResponse<Page<StoreDTO>> response = this.storeApiService.listStore(storeSearchDTO, 1, 1);
if(!response.isSuccess()){
continue;
}
if(CollectionUtils.isNotEmpty(response.getResult().getResult())){
StoreDTO storeDTO = new StoreDTO();
storeDTO.setStoreInfoId(storeInfoId);
storeDTO.setEnterpriseId(enterpriseId);
storeDTO.setFromEnterpriseId(dto.getUnionEnterpriseId());
this.storeAuthorizationApiService.saveStore(storeDTO);
}else {
this.storeAuthorizationApiService.cancelStore(storeInfoId, dto.getUnionEnterpriseId());
}
if(Integer.valueOf(fieldMap.get("old_region_id").getValue()) != Integer.valueOf(fieldMap.get("new_region_id").getValue())){
StoreSearchDTO storeSearchDTO1 = new StoreSearchDTO();
storeSearchDTO1.setRegionId(Integer.valueOf(fieldMap.get("old_region_id").getValue()));
storeSearchDTO1.setSearchJson(storeWidgetDTO.getSearchParam());
storeSearchDTO1.setEnterpriseId(enterpriseId);
ServiceResponse<Page<StoreDTO>> response1 = this.storeApiService.listStore(storeSearchDTO, 1, 1);
if(CollectionUtils.isEmpty(response1.getResult().getResult())){
this.storeRegionApiService.unAuthorizeRegion(dto.getUnionEnterpriseId(), Integer.valueOf(fieldMap.get("old_region_id").getValue()));
}
}
}
}
}
}
}
}
src/main/java/com/gic/message/service/impl/StoreMessageHandler.java
......@@ -51,62 +51,69 @@ public class StoreMessageHandler implements MessageHandler {
this.storeIndexRefreshHandler.refreshStoreIndex(enterpriseId, storeId);
Integer ownType = Integer.valueOf(fieldMap.get("own_type").getValue());
if(ownType == StoreOwnTypeEnum.OWNER.getCode()){
ServiceResponse<List<UnionEnterpriseResourceDTO>> resourceResponse = this.unionEnterpriseApiService.listStoreResourceByEnterpriseId(enterpriseId);
if(CollectionUtils.isNotEmpty(resourceResponse.getResult())){
for(UnionEnterpriseResourceDTO dto : resourceResponse.getResult()){
ServiceResponse<StoreWidgetDTO> storeWidget = this.storeWidgetApiService.getStoreWidget(dto.getResource().intValue());
if(!storeWidget.isSuccess()){
continue;
}
StoreWidgetDTO storeWidgetDTO = storeWidget.getResult();
if(storeWidgetDTO != null){
if(AuthModeEnum.YES.getCode() == storeWidgetDTO.getAuthMode()){
StoreSearchDTO storeSearchDTO = new StoreSearchDTO();
storeSearchDTO.setStoreIds(String.valueOf(storeId));
storeSearchDTO.setSearchJson(storeWidgetDTO.getSearchParam());
storeSearchDTO.setEnterpriseId(enterpriseId);
ServiceResponse<Page<StoreDTO>> response = this.storeApiService.listStore(storeSearchDTO, 1, 1);
if(!response.isSuccess()){
continue;
}
if(CollectionUtils.isNotEmpty(response.getResult().getResult())){
StoreDTO storeDTO = new StoreDTO();
storeDTO.setStoreInfoId(storeInfoId);
storeDTO.setEnterpriseId(enterpriseId);
storeDTO.setFromEnterpriseId(dto.getUnionEnterpriseId());
this.storeAuthorizationApiService.saveStore(storeDTO);
}
}
}
}
}
this.dealStoreChange(enterpriseId, storeId, storeInfoId, GicRecordType.INSERT.value());
}
}
if(GicRecordType.UPDATE.value() == value.getRecordType().value()){
int isAdd = 0;
if(Integer.valueOf(fieldMap.get("old_status").getValue()) == 0 && Integer.valueOf(fieldMap.get("new_status").getValue()) == 1){
isAdd = 1;
}
if(Integer.valueOf(fieldMap.get("old_status").getValue()) == 1 && Integer.valueOf(fieldMap.get("new_status").getValue()) == 0){
isAdd = 2;
}
if(GicRecordType.UPDATE.value() == value.getRecordType().value()) {
enterpriseId = Integer.valueOf(fieldMap.get("new_enterprise_id").getValue());
storeId = Integer.valueOf(fieldMap.get("new_store_id").getValue());
if(isAdd != 0){
if(isAdd == 1){
this.storeIndexRefreshHandler.refreshStoreIndex(enterpriseId, storeId);
}else {
this.storeApiService.deleteStoreFromES(enterpriseId, storeId);
}
}else {
this.storeIndexRefreshHandler.refreshStoreIndex(enterpriseId, storeId);
}
storeInfoId = Integer.valueOf(fieldMap.get("new_store_info_id").getValue());
if (Integer.valueOf(fieldMap.get("old_status").getValue()) == 1 && Integer.valueOf(fieldMap.get("new_status").getValue()) == 0) {
this.storeApiService.deleteStoreFromES(enterpriseId, storeId);
} else {
this.storeIndexRefreshHandler.refreshStoreIndex(enterpriseId, storeId);
}
if (StoreOwnTypeEnum.OWNER.getCode() == Integer.valueOf(fieldMap.get("new_own_type").getValue())) {
this.dealStoreChange(enterpriseId, storeId, storeInfoId, GicRecordType.UPDATE.value());
}
}
if(GicRecordType.DELETE.value() == value.getRecordType().value()){
enterpriseId = Integer.valueOf(fieldMap.get("enterprise_id").getValue());
storeId = Integer.valueOf(fieldMap.get("store_id").getValue());
storeInfoId = Integer.valueOf(fieldMap.get("store_info_id").getValue());
this.storeApiService.deleteStoreFromES(enterpriseId, storeId);
Integer ownType = Integer.valueOf(fieldMap.get("own_type").getValue());
if(ownType == StoreOwnTypeEnum.OWNER.getCode()){
this.dealStoreChange(enterpriseId, storeId, storeInfoId, GicRecordType.DELETE.value());
}
}
}
private void dealStoreChange(Integer enterpriseId, Integer storeId, Integer storeInfoId, int recordType){
ServiceResponse<List<UnionEnterpriseResourceDTO>> resourceResponse = this.unionEnterpriseApiService.listStoreResourceByEnterpriseId(enterpriseId);
if(CollectionUtils.isNotEmpty(resourceResponse.getResult())){
for(UnionEnterpriseResourceDTO dto : resourceResponse.getResult()){
ServiceResponse<StoreWidgetDTO> storeWidget = this.storeWidgetApiService.getStoreWidget(dto.getResource().intValue());
if(!storeWidget.isSuccess()){
continue;
}
StoreWidgetDTO storeWidgetDTO = storeWidget.getResult();
if(storeWidgetDTO != null){
if(AuthModeEnum.YES.getCode() == storeWidgetDTO.getAuthMode()){
StoreSearchDTO storeSearchDTO = new StoreSearchDTO();
storeSearchDTO.setStoreIds(String.valueOf(storeId));
storeSearchDTO.setSearchJson(storeWidgetDTO.getSearchParam());
storeSearchDTO.setEnterpriseId(enterpriseId);
ServiceResponse<Page<StoreDTO>> response = this.storeApiService.listStore(storeSearchDTO, 1, 1);
if(!response.isSuccess()){
continue;
}
if(CollectionUtils.isNotEmpty(response.getResult().getResult())){
if(GicRecordType.DELETE.value() == recordType){
this.storeAuthorizationApiService.cancelStore(storeInfoId, dto.getUnionEnterpriseId());
}else {
StoreDTO storeDTO = new StoreDTO();
storeDTO.setStoreInfoId(storeInfoId);
storeDTO.setEnterpriseId(enterpriseId);
storeDTO.setFromEnterpriseId(dto.getUnionEnterpriseId());
this.storeAuthorizationApiService.saveStore(storeDTO);
}
}else {
this.storeAuthorizationApiService.cancelStore(storeInfoId, dto.getUnionEnterpriseId());
}
}
}
}
}
}
}
src/main/resources/dubbo-gic-platform-message.xml
......@@ -23,4 +23,6 @@
<dubbo:reference interface="com.gic.auth.service.UnionEnterpriseApiService" id="unionEnterpriseApiService" timeout="6000" />
<dubbo:reference interface="com.gic.store.service.StoreWidgetApiService" id="storeWidgetApiService" timeout="6000" />
<dubbo:reference interface="com.gic.store.service.StoreAuthorizationApiService" id="storeAuthorizationApiService" timeout="6000" />
<dubbo:reference interface="com.gic.store.service.StoreRegionApiService" id="storeRegionApiService" timeout="6000" />
</beans>
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment