io.eventuate.tram.messaging.common.Message类的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(11.0k)|赞(0)|评价(0)|浏览(153)

本文整理了Java中io.eventuate.tram.messaging.common.Message类的一些代码示例，展示了Message类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Message类的具体详情如下：
包路径：io.eventuate.tram.messaging.common.Message
类名称：Message

Message介绍

[英]A message
[中]留言

代码示例

代码示例来源：origin: eventuate-tram/eventuate-tram-sagas

private String getSagaType(Message message) {
 return message.getRequiredHeader(SagaCommandHeaders.SAGA_TYPE);
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

private void send(Message message) {
 String table = eventuateSchema.qualifyTable("message");
 jdbcTemplate.update(String.format("insert into %s(id, destination, headers, payload) values(?, ?, ?, ?)", table),
     message.getId(),
     message.getRequiredHeader(Message.DESTINATION),
     JSonMapper.toJson(message.getHeaders()),
     message.getPayload());
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

protected void sendMessage(String id, String destination, Message message, MessageSender messageSender) {
 if (id == null) {
  if (!message.getHeader(Message.ID).isPresent())
   throw new IllegalArgumentException("message needs an id");
 } else {
  message.getHeaders().put(Message.ID, id);
 }
 message.getHeaders().put(Message.DESTINATION, destination);
 message.getHeaders().put(Message.DATE, HttpDateHeaderFormatUtil.nowAsHttpDateString());
 preSend(message);
 try {
  messageSender.send(message);
  postSend(message, null);
 } catch (RuntimeException e) {
  postSend(message, e);
  throw e;
 }
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

public MessageBuilder(Message message) {
 this(message.getPayload());
 this.headers = message.getHeaders();
}

代码示例来源：origin: eventuate-tram/eventuate-tram-sagas

@Override
public void stashMessage(String sagaType, String sagaId, String target, Message message) {
 logger.debug("Stashing message from {} for {} : {}", sagaId, target, message);
 jdbcTemplate.update(insertIntoSagaStashTableSql,
     message.getRequiredHeader(Message.ID),
     target,
     sagaType,
     sagaId,
     JSonMapper.toJson(message.getHeaders()),
     message.getPayload()
     );
}

代码示例来源：origin: io.eventuate.tram.sagas/eventuate-jpa-sagas-framework

String sagaId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_ID);
String sagaType = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_TYPE);
String requestId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_REQUEST_ID);
String messageId = message.getId();
String messageType = message.getRequiredHeader(ReplyMessageHeaders.REPLY_TYPE);
String messageJson = message.getPayload();
message.getHeader(SagaReplyHeaders.REPLY_LOCKED).ifPresent(lockedTarget -> {
 String destination = message.getRequiredHeader(CommandMessageHeaders.inReply(CommandMessageHeaders.DESTINATION));
 sagaInstance.addDestinationsAndResources(singleton(new DestinationAndResource(destination, lockedTarget)));
});

代码示例来源：origin: eventuate-tram/eventuate-tram-core

public void messageHandler(Message message) {
 logger.trace("Received message {} {}", commandDispatcherId, message);
 Optional<CommandHandler> possibleMethod = commandHandlers.findTargetMethod(message);
 if (!possibleMethod.isPresent()) {
  throw new RuntimeException("No method for " + message);
 }
 CommandHandler m = possibleMethod.get();
 Object param = convertPayload(m, message.getPayload());
 Map<String, String> correlationHeaders = correlationHeaders(message.getHeaders());
 Map<String, String> pathVars = getPathVars(message, m);
 Optional<String> defaultReplyChannel = message.getHeader(CommandMessageHeaders.REPLY_TO);
 List<Message> replies;
 try {
  CommandMessage cm = new CommandMessage(message.getId(), param, correlationHeaders, message);
  replies = invoke(m, cm, pathVars);
  logger.trace("Generated replies {} {} {}", commandDispatcherId, message, replies);
 } catch (Exception e) {
  logger.trace("Generated error {} {} {}", commandDispatcherId, message, e.getClass().getName());
  handleException(message, param, m, e, pathVars, defaultReplyChannel);
  return;
 }
 if (replies != null) {
  publish(correlationHeaders, replies, defaultReplyChannel);
 } else {
  logger.trace("Null replies - not publishling");
 }
}

代码示例来源：origin: io.eventuate.tram.core/eventuate-tram-events

public void messageHandler(Message message) {
 String aggregateType = message.getRequiredHeader(EventMessageHeaders.AGGREGATE_TYPE);
 Optional<DomainEventHandler> handler = domainEventHandlers.findTargetMethod(message);
 if (!handler.isPresent()) {
  return;
 }
 DomainEvent param = JSonMapper.fromJson(message.getPayload(), handler.get().getEventClass());
 handler.get().invoke(new DomainEventEnvelopeImpl<>(message,
     aggregateType,
     message.getRequiredHeader(EventMessageHeaders.AGGREGATE_ID),
     message.getRequiredHeader(Message.ID),
     param));
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

public DomainEventHandlerUnitTestSupport publishes(DomainEvent event) {
 DomainEventPublisher publisher = new DomainEventPublisherImpl((destination, message) -> {
  String id = idGenerator.generateId().toString();
  message.getHeaders().put(Message.ID, id);
  handler.accept(message);
 });
 publisher.publish(aggregateType, aggregateId, Collections.singletonList(event));
 return this;
}

代码示例来源：origin: eventuate-tram/eventuate-tram-sagas

public <C extends Command> SagaParticipantStubManagerHelper<C> when(C expectedCommand) {
 return new SagaParticipantStubManagerHelper<C>(this, (Class<C>) expectedCommand.getClass(),
     message -> JSonMapper.fromJson(message.getPayload(), expectedCommand.getClass()).equals(expectedCommand));
}

代码示例来源：origin: io.eventuate.tram.core/eventuate-tram-consumer-rabbitmq

private void handleMessage(String subscriberId, MessageHandler handler, Message tramMessage, Runnable acknowledgeCallback) {
 transactionTemplate.execute(ts -> {
  if (duplicateMessageDetector.isDuplicate(subscriberId, tramMessage.getId())) {
   logger.info("consumer {} with subscriberId {} received message duplicate with id{}", id, subscriberId, tramMessage.getId());
   acknowledgeCallback.run();
   return null;
  }
  try {
   handler.accept(tramMessage);
   logger.info("consumer {} with subscriberId {} handled message with id {}", id, subscriberId, tramMessage.getId());
  } catch (Throwable t) {
   logger.info("consumer {} with subscriberId {} got exception when tried to handle message with id {}", id, subscriberId, tramMessage.getId());
   logger.info("Got exception ", t);
  } finally {
   acknowledgeCallback.run();
  }
  return null;
 });
}

代码示例来源：origin: eventuate-tram/eventuate-tram-sagas

private void handleReply(Message message) {
 if (!isReplyForThisSagaType(message))
  return;
 logger.debug("Handle reply: {}", message);
 String sagaId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_ID);
 String sagaType = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_TYPE);
 SagaInstance sagaInstance = sagaInstanceRepository.find(sagaType, sagaId);
 Data sagaData = SagaDataSerde.deserializeSagaData(sagaInstance.getSerializedSagaData());
 message.getHeader(SagaReplyHeaders.REPLY_LOCKED).ifPresent(lockedTarget -> {
  String destination = message.getRequiredHeader(CommandMessageHeaders.inReply(CommandMessageHeaders.DESTINATION));
  sagaInstance.addDestinationsAndResources(singleton(new DestinationAndResource(destination, lockedTarget)));
 });
 String currentState = sagaInstance.getStateName();
 logger.info("Current state={}", currentState);
 SagaActions<Data> actions = getStateDefinition().handleReply(currentState, sagaData, message);
 logger.info("Handled reply. Sending commands {}", actions.getCommands());
 processActions(sagaId, sagaInstance, sagaData, actions);
}

代码示例来源：origin: eventuate-tram/eventuate-tram-sagas

@Test
public void shouldReleaseLockAndUnstashMessage() {
 String sagaId1 = idGenerator.genId().toString();
 String sagaId2 = idGenerator.genId().toString();
 String target  = "/target/" + idGenerator.genId().toString();
 String messageId = idGenerator.genId().toString();
 assertTrue(sagaLockManager.claimLock(sagaType, sagaId1, target));
 assertFalse(sagaLockManager.claimLock(sagaType, sagaId2, target));
 String payload = "hello";
 Message message = MessageBuilder.withPayload(payload).withHeader(Message.ID, messageId).build();
 sagaLockManager.stashMessage(sagaType, sagaId2, target, message);
 Optional<Message> unstashedMessage1 = sagaLockManager.unlock(sagaId1, target);
 assertTrue(unstashedMessage1.isPresent());
 assertEquals(messageId, unstashedMessage1.get().getId());
 assertEquals(payload, unstashedMessage1.get().getPayload());
 assertFalse(sagaLockManager.unlock(sagaId2, target).isPresent());
}

代码示例来源：origin: io.eventuate.tram.sagas/eventuate-jpa-sagas-framework

public void handleMessage(Message message) {
 logger.debug("handle message invoked {}", message);
 if (message.hasHeader(SagaReplyHeaders.REPLY_SAGA_ID)) {
  handleReply(message);
 } else if (message.hasHeader(EventMessageHeaders.EVENT_TYPE)) {
  String aggregateType = message.getRequiredHeader(EventMessageHeaders.AGGREGATE_TYPE);
  String aggregateId = message.getRequiredHeader(Message.PARTITION_ID);
  String eventType = message.getRequiredHeader(EventMessageHeaders.EVENT_TYPE);
  // TODO query the saga event routing table: (at, aId, et) -> [(sagaType, sagaId)]
  for (SagaTypeAndId sagaTypeAndId : aggregateInstanceSubscriptionsDAO.findSagas(aggregateType, aggregateId, eventType)) {
   handleAggregateInstanceEvent(sagaTypeAndId.getSagaType(), sagaTypeAndId.getSagaId(), message, aggregateType, aggregateId, eventType);
  }
  ;
 } else {
  logger.warn("Handle message doesn't know what to do with: {} ", message);
 }
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

@Override
public String get(String key) {
 return message.getHeader(key).orElse(null);
}

代码示例来源：origin: io.eventuate.tram.sagas/eventuate-jpa-sagas-framework

@Override
public void stashMessage(String sagaType, String sagaId, String target, Message message) {
 logger.debug("Stashing message from {} for {} : {}", sagaId, target, message);
 jdbcTemplate.update(insertIntoSagaStashTableSql,
     message.getRequiredHeader(Message.ID),
     target,
     sagaType,
     sagaId,
     JSonMapper.toJson(message.getHeaders()),
     message.getPayload()
     );
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

public void messageHandler(Message message) {
 String aggregateType = message.getRequiredHeader(EventMessageHeaders.AGGREGATE_TYPE);
 Optional<DomainEventHandler> handler = domainEventHandlers.findTargetMethod(message);
 if (!handler.isPresent()) {
  return;
 }
 DomainEvent param = JSonMapper.fromJson(message.getPayload(), handler.get().getEventClass());
 handler.get().invoke(new DomainEventEnvelopeImpl<>(message,
     aggregateType,
     message.getRequiredHeader(EventMessageHeaders.AGGREGATE_ID),
     message.getRequiredHeader(Message.ID),
     param));
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

private boolean headersMatch(Message message, Contract groovyDsl) {
  Map<String, String> headers = message.getHeaders();
  boolean matches = true;
  for (Header it : groovyDsl.getInput().getMessageHeaders().getEntries()) {
   String name = it.getName();
   Object value = it.getClientValue();
   Object valueInHeader = headers.get(name);
   matches &= value instanceof Pattern ?
       ((Pattern) value).matcher(valueInHeader.toString()).matches() :
       valueInHeader != null && valueInHeader.equals(value);
  }
  return matches;
 }
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

@Override
 protected ContractVerifierMessage convert(Message m) {
  return m == null ? null : contractVerifierMessaging.create(m.getPayload(), m.getHeaders());
 }
}

代码示例来源：origin: eventuate-tram/eventuate-tram-core

@Override
public void accept(Message message) {
 try {
  queue.put(message.getPayload());
 } catch (InterruptedException e) {
  throw new RuntimeException(e);
 }
}

内容来源于网络，如有侵权，请联系作者删除！

Message

相关文章

    查看更多
    微信公众号

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com