import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Bean
public SpringAMQPMessageSource messageSourceApiGateway(Serializer serializer) {
    return new SpringAMQPMessageSource(new DefaultAMQPMessageConverter(serializer)) {

        @RabbitListener(queues = "${spring.application.queue}")
        @Override
        public void onMessage(Message message, Channel channel) throws Exception {
            super.onMessage(message, channel);
        }
    };
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
public void appendEvents(Bucket bucket, List<? extends EventMessage<?>> events, Serializer serializer) {
    List<JsonObject> jsonObjects = createEventDocuments(events, serializer).collect(Collectors.toList());
    jsonObjects.forEach((e) -> {
        String docId = EVENT_PREFIX + e.getString("aggregateIdentifier");
        if (!bucket.exists(docId)) {
            JsonArray eventArray = JsonArray.empty()
                    .add(e);
            JsonObject data = JsonObject.empty()
                    .put("events", eventArray);
            JsonDocument doc = JsonDocument.create(docId, data);
            bucket.insert(doc);
        } else {
            bucket.mutateIn(docId)
                    .arrayAppend("events", e, false)
                    .execute();
        }

    });
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
/**
 * Constructor used to create a new event entry to store in Mongo.
 *
 * @param event The actual DomainEvent to store
 * @param serializer Serializer to use for the event to store
 */
EventEntry(DomainEventMessage<?> event, Serializer serializer) {
    this.aggregateType = event.getType();
    this.aggregateIdentifier = event.getAggregateIdentifier();
    this.sequenceNumber = event.getSequenceNumber();
    this.eventIdentifier = event.getIdentifier();
    Class<?> serializationTarget = String.class;
    if (serializer.canSerializeTo(JsonObject.class)) {
        serializationTarget = JsonObject.class;
    }
    SerializedObject<?> serializedPayloadObject = serializePayload(event, serializer, serializationTarget);
    SerializedObject<?> serializedMetaDataObject = serializeMetaData(event, serializer, serializationTarget);

    this.serializedPayload = serializedPayloadObject.getData();
    this.payloadType = serializedPayloadObject.getType().getName();
    this.payloadRevision = serializedPayloadObject.getType().getRevision();
    this.serializedMetaData = serializedMetaDataObject.getData();
    this.timeStamp = event.getTimestamp().toString();
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Bean
public SpringAMQPMessageSource myMessageSource(Serializer serializer, EventHandlingConfiguration ehConfig) {
  return new SpringAMQPMessageSource(serializer) {
    @RabbitListener(queues = "${camunda.bpm.cloud.amqp.queue.event}")
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
      log.info("receiving event: {}#{}", message, channel);
      super.onMessage(message, channel);
    }
  };
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
public CouchbaseEventStoreEngine(Serializer serializer,
        EventUpcaster upcasterChain,
        PersistenceExceptionResolver persistenceExceptionResolver,
        Integer batchSize,
        CouchbaseStoreStrategy storageStrategy,
        CouchbaseTemplate template) {
    super(serializer, upcasterChain, persistenceExceptionResolver, batchSize);
    this.storageStrategy = storageStrategy;
    this.template = template;
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
protected void appendEvents(List<? extends EventMessage<?>> events, Serializer serializer) {
    if (!events.isEmpty()) {
        try {
            storageStrategy.appendEvents(template.getEventBucket(), events, serializer);
        } catch (Exception e) {
            handlePersistenceException(e, events.get(0));
        }
    }
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
protected void storeSnapshot(DomainEventMessage<?> snapshot, Serializer serializer) {
    try {
        storageStrategy.appendSnapshot(template.getSnapshotBucket(), snapshot, serializer);
    } catch (Exception e) {
        handlePersistenceException(e, snapshot);
    }
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
public TokenEntry(String process, int segment, TrackingToken tokenBuffer, Serializer serializer) {
    SerializedObject<byte[]> serializedToken = serializer.serialize(tokenBuffer, byte[].class);
    this.processorName = process;
    this.segment = segment;
    this.tokenBuffer = ByteBuffer.wrap(serializedToken.getData());
    this.tokenType = serializedToken.getType().getName();
    this.timeStamp = new Date();
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
public CassandraTokenStore(Session session, Serializer serializer, TokenSchema schema) {
    if (session == null) {
        throw new IllegalArgumentException("Parameter 'session' cannot be null");
    }
    MappingManager mappingManager = new MappingManager(session);
    this.tokenMapper = mappingManager.mapper(TokenEntry.class);
    this.serializer = getOrDefault(serializer, XStreamSerializer::new);
    this.schema = getOrDefault(schema, TokenSchema.builder().build());
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
public CassandraEventStorageEngine(Serializer serializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, Integer batchSize, Session session, EventSchema schema) {
    super(serializer, upcasterChain, persistenceExceptionResolver, batchSize, session, schema);

    this.counterSelectStatement = session.prepare("SELECT " + quoted(schema().valueColumn()) +
            " FROM" + quoted(schema().countersTable()) +
            " WHERE " + quoted(schema().nameColumn()) + " = ? LIMIT 1");
    this.counterInsertStatement = session.prepare("INSERT INTO " + quoted(schema().countersTable()) +
            " (" + quoted(schema().nameColumn(), schema().valueColumn()) + ")" +
            " VALUES(?,?)");

    Row globalIndexCounterRow = session.execute(counterSelectStatement.bind(GLOBAL_INDEX_COUNTER_NAME)).one();
    this.globalIndexCounter = new AtomicLong(Optional.ofNullable(globalIndexCounterRow)
            .map(row -> row.getLong(0))
            .orElse(0L));
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
protected void appendEvents(List<? extends EventMessage<?>> events, Serializer serializer) {
    events.stream()
            .map(EventUtils::asDomainEventMessage)
            .map(e -> asDomainEventEntry(e, serializer, globalIndexCounter.getAndIncrement()))
            .map(this::storeEventLogEntry)
            .map(eventMapper::saveQuery)
            .forEachOrdered(batch()::add);
    batch().add(counterInsertStatement.bind(GLOBAL_INDEX_COUNTER_NAME, globalIndexCounter.get()));
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
public DefaultAxonMessageSerializer(Serializer delegate, EventUpcasterChain eventUpcasterChain) {
    this.delegate = new MessageSerializer(delegate);
    upcasterChain = eventUpcasterChain;
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
protected Stream<JsonObject> createEventDocuments(List<? extends EventMessage<?>> events, Serializer serializer) {
    return events.stream()
            .map(EventUtils::asDomainEventMessage)
            .map(event -> new EventEntry(event, serializer))
            .map(entry -> entry.asJsonObject());
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
public void appendSnapshot(Bucket bucket, DomainEventMessage<?> snapshot, Serializer serializer) {
    throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
public TrackingToken trackingToken(Serializer serializer) {
    SimpleSerializedObject<byte[]> serializedObject =
            new SimpleSerializedObject<>(tokenBuffer != null ? tokenBuffer.array() : null, byte[].class, new SimpleSerializedType(tokenType, null));
    return serializer.deserialize(serializedObject);
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
private static DomainEventEntry asDomainEventEntry(DomainEventMessage<?> eventMessage, Serializer serializer, long globalIndex) {
    return new DomainEventEntry(globalIndex, eventMessage, serializer);
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
private static SnapshotEventEntry asSnapshotEventEntry(DomainEventMessage<?> eventMessage, Serializer serializer) {
    return new SnapshotEventEntry(eventMessage, serializer);
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
protected void storeSnapshot(DomainEventMessage<?> snapshot, Serializer serializer) {
    batch().add(snapshotMapper.saveQuery(asSnapshotEventEntry(snapshot, serializer)));
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
protected void appendEvents(List<? extends EventMessage<?>> events, Serializer serializer) {
    throw new UnsupportedOperationException();
}
 

import org.axonframework.serialization.Serializer; //导入依赖的package包/类
@Override
protected void storeSnapshot(DomainEventMessage<?> snapshot, Serializer serializer) {
    throw new UnsupportedOperationException();
}