Skip to content

张云鹏 / uc-consumer

Go to a project
Toggle navigation
Toggle navigation pinning
Commit b1f50657 b1f50657c931922e5f4c86a9df8f95acf592ef3c by xianghan
Options

1.兼容1.0.0版本

1 parent 1a39d208
Inline Side-by-side
Showing 7 changed files with 280 additions and 4 deletions
...@@ -80,4 +80,7 @@ public interface MemberService { ...@@ -80,4 +80,7 @@ public interface MemberService {
80 MemberDTO doUpdateMemberExpAndLevel(Member resources); 80 MemberDTO doUpdateMemberExpAndLevel(Member resources);
81 81
82 MemberDTO unbindUserIpTv(Member member); 82 MemberDTO unbindUserIpTv(Member member);
83
84
85 void unbind(Member resources);
83 } 86 }
......
...@@ -147,6 +147,27 @@ public class MemberServiceImpl implements MemberService { ...@@ -147,6 +147,27 @@ public class MemberServiceImpl implements MemberService {
147 147
148 @Override 148 @Override
149 @Transactional(rollbackFor = Exception.class) 149 @Transactional(rollbackFor = Exception.class)
150 public void unbind(Member resources) {
151 try {
152 String code = resources.getCode();
153 MemberDTO memberDTO = this.findByCode(code);
154 // Member member = memberRepository.findById(resources.getId()).orElseGet(Member::new);
155 ValidationUtil.isNull(memberDTO.getId(), "Member", "id", resources.getId());
156 Member member = new Member();
157 BeanUtils.copyProperties(memberDTO,member);
158
159 member.setUserIptvId(null);
160 member.setBindIptvTime(null);
161 member.setBindIptvPlatformType(null);
162 this.save(member);
163 } catch (Exception e) {
164 e.printStackTrace();
165 throw e;
166 }
167 }
168
169 @Override
170 @Transactional(rollbackFor = Exception.class)
150 public MemberDTO update(Member resources) { 171 public MemberDTO update(Member resources) {
151 172
152 log.info("MemberServiceImpl ==>> update ==>> resources ==>> [{}]" , resources); 173 log.info("MemberServiceImpl ==>> update ==>> resources ==>> [{}]" , resources);
...@@ -201,4 +222,6 @@ public class MemberServiceImpl implements MemberService { ...@@ -201,4 +222,6 @@ public class MemberServiceImpl implements MemberService {
201 222
202 } 223 }
203 224
225
226
204 } 227 }
......
...@@ -19,6 +19,7 @@ import org.springframework.transaction.annotation.Propagation; ...@@ -19,6 +19,7 @@ import org.springframework.transaction.annotation.Propagation;
19 import org.springframework.transaction.annotation.Transactional; 19 import org.springframework.transaction.annotation.Transactional;
20 import org.springframework.util.Assert; 20 import org.springframework.util.Assert;
21 21
22 import javax.validation.constraints.NotNull;
22 import java.util.Objects; 23 import java.util.Objects;
23 import java.util.Optional; 24 import java.util.Optional;
24 25
...@@ -120,6 +121,13 @@ public class UserTvServiceImpl implements UserTvService { ...@@ -120,6 +121,13 @@ public class UserTvServiceImpl implements UserTvService {
120 @Override 121 @Override
121 @Transactional(rollbackFor = Exception.class) 122 @Transactional(rollbackFor = Exception.class)
122 public void unbindPriorityMemberCode(UserTv resources) { 123 public void unbindPriorityMemberCode(UserTv resources) {
124 String platformAccount = resources.getPlatformAccount();
125 if (StringUtils.isNotBlank(platformAccount)) {
126 UserTvDTO userTvDTO = this.findByPlatformAccount(platformAccount);
127 Long id = userTvDTO.getId();
128 resources.setId(id);
129 resources.setMemberId(userTvDTO.getMemberId());
130 }
123 this.userTvRepository.save(resources); 131 this.userTvRepository.save(resources);
124 } 132 }
125 133
......
...@@ -226,6 +226,18 @@ public class RabbitMqSourceConfig { ...@@ -226,6 +226,18 @@ public class RabbitMqSourceConfig {
226 return COLLECTION_QUEUE; 226 return COLLECTION_QUEUE;
227 } 227 }
228 228
229 public String getUcgCollectionQueueAdd(){
230 return "queue.collection.add";
231 }
232
233 public String getUcgCollectionQueueDelete(){
234 return "queue.collection.delete";
235 }
236
237 public String getUcgCollectionQueueDeleteAll(){
238 return "queue.collection.deleteall";
239 }
240
229 public String getUcgCollectionSource(){ 241 public String getUcgCollectionSource(){
230 if (Objects.nonNull(ucgIptvCollectionInfo)) { 242 if (Objects.nonNull(ucgIptvCollectionInfo)) {
231 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) { 243 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) {
...@@ -346,6 +358,12 @@ public class RabbitMqSourceConfig { ...@@ -346,6 +358,12 @@ public class RabbitMqSourceConfig {
346 return "false"; 358 return "false";
347 } 359 }
348 360
361
362
363 public String getMemberInfoAsyncQueue(){
364 return "queue.MemberInfoSync";
365 }
366
349 /**************************************************eventBus*************************************************************/ 367 /**************************************************eventBus*************************************************************/
350 public static final String UC_EVENTBUS_EXCHANGE = "uc.eventbus"; 368 public static final String UC_EVENTBUS_EXCHANGE = "uc.eventbus";
351 public static final String UC_EVENTBUS_KEY = "uc.eventbus.*.topic"; 369 public static final String UC_EVENTBUS_KEY = "uc.eventbus.*.topic";
......
...@@ -31,6 +31,43 @@ public class UcEngineManagement2IptvConsumer { ...@@ -31,6 +31,43 @@ public class UcEngineManagement2IptvConsumer {
31 @Value("#{rabbitMqErrorLogConfig.getUceError()}") 31 @Value("#{rabbitMqErrorLogConfig.getUceError()}")
32 private Map<String, String> error; 32 private Map<String, String> error;
33 33
34 @RabbitHandler
35 @RabbitListener(queues = "#{rabbitMqSourceConfig.getMemberInfoAsyncQueue()}",
36 containerFactory = "#{rabbitMqSourceConfig.getUceSource()}",
37 autoStartup = "#{rabbitMqSourceConfig.getUceStartUp()}",
38 ackMode = "AUTO")
39 public void ucEventConsumer2(Channel channel, Message message, String content) throws IOException {
40 log.info(" receive MemberInfoAsync msg , content is : {} ", content);
41
42 try {
43 TableOperationMsg tableOperationMsg = this.parseContent(content);
44
45 autoUser.route(tableOperationMsg);
46
47 // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
48
49 } catch (Exception e) {
50
51 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
52
53 if (MapUtils.isNotEmpty(error)) {
54 String errorStart = this.error.get("start");
55
56 if (errorStart.equalsIgnoreCase("true")) {
57 String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
58 String filePath = this.error.get("filePath");
59 String filePath1 = filePath+fileName;
60 FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
61 }
62
63 }
64
65 e.printStackTrace();
66 }
67
68 log.info("ucEventConsumer ====>>>> end");
69 }
70
34 /** 71 /**
35 * 事件 72 * 事件
36 * @param content 73 * @param content
...@@ -44,7 +81,7 @@ public class UcEngineManagement2IptvConsumer { ...@@ -44,7 +81,7 @@ public class UcEngineManagement2IptvConsumer {
44 autoStartup = "#{rabbitMqSourceConfig.getUceStartUp()}", 81 autoStartup = "#{rabbitMqSourceConfig.getUceStartUp()}",
45 ackMode = "MANUAL") 82 ackMode = "MANUAL")
46 public void ucEventConsumer(Channel channel, Message message, String content) throws IOException { 83 public void ucEventConsumer(Channel channel, Message message, String content) throws IOException {
47 log.info(" receive dataSync msg , content is : {} ", content); 84 log.info(" receive ucEventConsumer msg , content is : {} ", content);
48 85
49 try { 86 try {
50 TableOperationMsg tableOperationMsg = this.parseContent(content); 87 TableOperationMsg tableOperationMsg = this.parseContent(content);
......
...@@ -239,4 +239,185 @@ public class UcGatewayIptv2IptvConsumer { ...@@ -239,4 +239,185 @@ public class UcGatewayIptv2IptvConsumer {
239 e.printStackTrace(); 239 e.printStackTrace();
240 } 240 }
241 } 241 }
242
243
244
245
246
247 /**
248 * @description 添加收藏记录
249 * @param content 消息内容
250 */
251 @RabbitHandler
252 @RabbitListener(queues = "#{rabbitMqSourceConfig.getUcgCollectionQueueAdd()}",
253 containerFactory = "#{rabbitMqSourceConfig.getUcgCollectionSource()}",
254 autoStartup = "#{rabbitMqSourceConfig.getUcgCollectionStartUp()}",
255 ackMode = "MANUAL")
256 public void collectionConsumerAdd(Channel channel, Message message, String content) throws IOException {
257 log.info("receive collectionConsumerAdd add message, content {}", content);
258
259 try {
260
261 JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
262 if (Objects.nonNull(content)) {
263 String evt = jsonObject.get("evt").toString();
264 String msgData = jsonObject.get("msgData").toString();
265 switch (evt.toUpperCase()) {
266 // 添加收藏
267 case "ADDCOLLECTION":
268 this.restTemplateClient.addCollection(msgData);
269 break;
270 // 删除收藏
271 case "DELETECOLLECTION":
272 this.restTemplateClient.deleteCollection(msgData);
273 break;
274 // 删除全部收藏
275 case "DELETEALLCOLLECTION":
276 this.restTemplateClient.deleteAllCollection(msgData);
277 break;
278 default:
279 break;
280
281 }
282 }
283
284 // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
285
286 } catch (Exception e) {
287
288 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
289
290 if (MapUtils.isNotEmpty(error)) {
291 String errorStart = this.error.get("start");
292
293 if (errorStart.equalsIgnoreCase("true")) {
294 String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
295 String filePath = this.error.get("filePath");
296 String filePath1 = filePath+fileName;
297 FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
298 }
299
300 }
301
302 e.printStackTrace();
303 }
304 }
305
306 /**
307 * @description 添加收藏记录
308 * @param content 消息内容
309 */
310 @RabbitHandler
311 @RabbitListener(queues = "#{rabbitMqSourceConfig.getUcgCollectionQueueDelete()}",
312 containerFactory = "#{rabbitMqSourceConfig.getUcgCollectionSource()}",
313 autoStartup = "#{rabbitMqSourceConfig.getUcgCollectionStartUp()}",
314 ackMode = "MANUAL")
315 public void collectionConsumerDelete(Channel channel, Message message, String content) throws IOException {
316 log.info("receive collectionConsumerDelete add message, content {}", content);
317
318 try {
319
320 JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
321 if (Objects.nonNull(content)) {
322 String evt = jsonObject.get("evt").toString();
323 String msgData = jsonObject.get("msgData").toString();
324 switch (evt.toUpperCase()) {
325 // 添加收藏
326 case "ADDCOLLECTION":
327 this.restTemplateClient.addCollection(msgData);
328 break;
329 // 删除收藏
330 case "DELETECOLLECTION":
331 this.restTemplateClient.deleteCollection(msgData);
332 break;
333 // 删除全部收藏
334 case "DELETEALLCOLLECTION":
335 this.restTemplateClient.deleteAllCollection(msgData);
336 break;
337 default:
338 break;
339
340 }
341 }
342
343 // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
344
345 } catch (Exception e) {
346
347 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
348
349 if (MapUtils.isNotEmpty(error)) {
350 String errorStart = this.error.get("start");
351
352 if (errorStart.equalsIgnoreCase("true")) {
353 String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
354 String filePath = this.error.get("filePath");
355 String filePath1 = filePath+fileName;
356 FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
357 }
358
359 }
360
361 e.printStackTrace();
362 }
363 }
364
365 /**
366 * @description 添加收藏记录
367 * @param content 消息内容
368 */
369 @RabbitHandler
370 @RabbitListener(queues = "#{rabbitMqSourceConfig.getUcgCollectionQueueDeleteAll()}",
371 containerFactory = "#{rabbitMqSourceConfig.getUcgCollectionSource()}",
372 autoStartup = "#{rabbitMqSourceConfig.getUcgCollectionStartUp()}",
373 ackMode = "MANUAL")
374 public void collectionConsumerDeleteAll(Channel channel, Message message, String content) throws IOException {
375 log.info("receive collectionConsumerDeleteAll add message, content {}", content);
376
377 try {
378
379 JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
380 if (Objects.nonNull(content)) {
381 String evt = jsonObject.get("evt").toString();
382 String msgData = jsonObject.get("msgData").toString();
383 switch (evt.toUpperCase()) {
384 // 添加收藏
385 case "ADDCOLLECTION":
386 this.restTemplateClient.addCollection(msgData);
387 break;
388 // 删除收藏
389 case "DELETECOLLECTION":
390 this.restTemplateClient.deleteCollection(msgData);
391 break;
392 // 删除全部收藏
393 case "DELETEALLCOLLECTION":
394 this.restTemplateClient.deleteAllCollection(msgData);
395 break;
396 default:
397 break;
398
399 }
400 }
401
402 // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
403
404 } catch (Exception e) {
405
406 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
407
408 if (MapUtils.isNotEmpty(error)) {
409 String errorStart = this.error.get("start");
410
411 if (errorStart.equalsIgnoreCase("true")) {
412 String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
413 String filePath = this.error.get("filePath");
414 String filePath1 = filePath+fileName;
415 FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
416 }
417
418 }
419
420 e.printStackTrace();
421 }
422 }
242 } 423 }
......
...@@ -142,11 +142,17 @@ service: ...@@ -142,11 +142,17 @@ service:
142 # routing-key: uc.eventbus.*.topic 142 # routing-key: uc.eventbus.*.topic
143 # active: service 143 # active: service
144 - source: uce 144 - source: uce
145 exchange: uce.exchange 145 exchange: uc.direct
146 queue: uce.queue 146 queue: uc.route.key.direct.event.bbb
147 exchange-type: direct 147 exchange-type: direct
148 routing-key: 148 routing-key:
149 active: management 149 active: service
150 - source: uce
151 exchange: exchange.MemberInfoSync
152 queue: queue.MemberInfoSync
153 exchange-type: direct
154 routing-key:
155 active: service
150 # - source: wechat 156 # - source: wechat
151 # exchange: weixin.subOrUnSub.direct 157 # exchange: weixin.subOrUnSub.direct
152 # queue: weixin.subOrUnSub.queue 158 # queue: weixin.subOrUnSub.queue
......