Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ public void handleCommandRecord(CommandRecord commandRecord,
}
} catch (Throwable t) {
// TODO: potentially see if we need to terminate the runtime
log.error("Exception while handling command, sending CommandExecutionError", t);
log.error("Exception while handling command {}:{}, sending CommandExecutionError", commandRecord.name(), commandRecord.version(), t);
commandExecutionError(commandRecord, protocolRecordConsumer, t);
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.elasticsoftware.akces.processmanager.AkcesProcess;
import org.elasticsoftware.cryptotrading.aggregates.orders.commands.RejectOrderCommand;
import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket;
import org.elasticsoftware.akces.events.DomainEvent;
import org.elasticsoftware.cryptotrading.aggregates.orders.events.BuyOrderRejectedEvent;
import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InsufficientFundsErrorEvent;
import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InvalidCryptoCurrencyErrorEvent;
Expand All @@ -30,9 +31,10 @@

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes({
@JsonSubTypes.Type(value = BuyOrderProcess.class, name = "BUY")
@JsonSubTypes.Type(value = BuyOrderProcess.class, name = "BUY"),
@JsonSubTypes.Type(value = SellOrderProcess.class, name = "SELL")
})
public sealed interface OrderProcess extends AkcesProcess permits BuyOrderProcess {
public sealed interface OrderProcess extends AkcesProcess permits BuyOrderProcess, SellOrderProcess {
String orderId();

CryptoMarket market();
Expand All @@ -45,11 +47,11 @@ public sealed interface OrderProcess extends AkcesProcess permits BuyOrderProces

OrderProcessState state();

BuyOrderRejectedEvent handle(InsufficientFundsErrorEvent error);
DomainEvent handle(InsufficientFundsErrorEvent error);

BuyOrderRejectedEvent handle(InvalidCryptoCurrencyErrorEvent error);
DomainEvent handle(InvalidCryptoCurrencyErrorEvent error);

BuyOrderRejectedEvent handle(RejectOrderCommand command);
DomainEvent handle(RejectOrderCommand command);

OrderProcess withState(OrderProcessState state);
}
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,9 @@
import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.MarketOrderFilledEvent;
import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.MarketOrderRejectedErrorEvent;
import org.elasticsoftware.cryptotrading.aggregates.orders.commands.FillBuyOrderCommand;
import org.elasticsoftware.cryptotrading.aggregates.orders.commands.FillSellOrderCommand;
import org.elasticsoftware.cryptotrading.aggregates.orders.commands.PlaceBuyOrderCommand;
import org.elasticsoftware.cryptotrading.aggregates.orders.commands.PlaceSellOrderCommand;
import org.elasticsoftware.cryptotrading.aggregates.orders.commands.RejectOrderCommand;
import org.elasticsoftware.cryptotrading.aggregates.orders.events.*;
import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.CancelReservationCommand;
Expand Down Expand Up @@ -108,6 +110,40 @@ public OrderProcessManagerState handle(BuyOrderFilledEvent event, OrderProcessMa
}});
}

@EventSourcingHandler
public OrderProcessManagerState handle(SellOrderCreatedEvent event, OrderProcessManagerState state) {
return new OrderProcessManagerState(state.userId(), new ArrayList<>(state.runningProcesses()) {{
add(new SellOrderProcess(
event.orderId(),
event.market(),
event.quantity(),
event.clientReference()));
}});
}

@EventSourcingHandler
public OrderProcessManagerState handle(SellOrderRejectedEvent event, OrderProcessManagerState state) {
return new OrderProcessManagerState(state.userId(), new ArrayList<>(state.runningProcesses()) {{
removeIf(process -> process.orderId().equals(event.orderId()));
}});
}

@EventSourcingHandler
public OrderProcessManagerState handle(SellOrderPlacedEvent event, OrderProcessManagerState state) {
return new OrderProcessManagerState(state.userId(), new ArrayList<>(state.runningProcesses()) {{
replaceAll(process -> process.orderId().equals(event.orderId())
? process.withState(OrderProcessState.PLACED)
: process);
}});
}

@EventSourcingHandler
public OrderProcessManagerState handle(SellOrderFilledEvent event, OrderProcessManagerState state) {
return new OrderProcessManagerState(state.userId(), new ArrayList<>(state.runningProcesses()) {{
removeIf(process -> process.orderId().equals(event.orderId()));
}});
}

/**
* This is the entry point for the user to place a buy order
*
Expand Down Expand Up @@ -137,8 +173,8 @@ public Stream<BuyOrderCreatedEvent> placeBuyOrder(PlaceBuyOrderCommand command,
command.clientReference()));
}

@CommandHandler(produces = BuyOrderRejectedEvent.class, errors = {})
public Stream<BuyOrderRejectedEvent> rejectOrder(RejectOrderCommand command, OrderProcessManagerState state) {
@CommandHandler(produces = {BuyOrderRejectedEvent.class, SellOrderRejectedEvent.class}, errors = {})
public Stream<DomainEvent> rejectOrder(RejectOrderCommand command, OrderProcessManagerState state) {
log.info("CommandHandler: Rejecting order for userId={}, orderId={}", state.userId(), command.orderId());
if (state.hasAkcesProcess(command.orderId())) {
return Stream.of(state.getAkcesProcess(command.orderId()).handle(command));
Expand Down Expand Up @@ -186,7 +222,75 @@ public Stream<BuyOrderFilledEvent> fillOrder(FillBuyOrderCommand command, OrderP
return Stream.empty();
}

@EventHandler(produces = BuyOrderPlacedEvent.class, errors = {})
/**
* This is the entry point for the user to place a sell order
*
* @param command
* @param state
* @return
*/
@CommandHandler(produces = SellOrderCreatedEvent.class, errors = {})
public Stream<SellOrderCreatedEvent> placeSellOrder(PlaceSellOrderCommand command, OrderProcessManagerState state) {
log.info("CommandHandler: Placing sell order for userId={}, market={}, quantity={}",
state.userId(), command.market().id(), command.quantity());
// we need to reserve the base currency amount on wallet of the user
String orderId = UUID.randomUUID().toString();
log.info("CommandHandler: Generated orderId={} for sell order", orderId);
// send command to reserve the amount of the base currency
getCommandBus().send(new ReserveAmountCommand(
state.userId(),
command.market().baseCrypto(),
command.quantity(),
orderId));
// register the sell order process
return Stream.of(new SellOrderCreatedEvent(
state.userId(),
orderId,
command.market(),
command.quantity(),
command.clientReference()));
}

@CommandHandler(produces = SellOrderFilledEvent.class, errors = {})
public Stream<SellOrderFilledEvent> fillSellOrder(FillSellOrderCommand command, OrderProcessManagerState state) {
log.info("CommandHandler: Filling sell order for userId={}, orderId={}, baseCurrency={}, quoteCurrency={}, quantity={}, price={}",
command.userId(), command.orderId(), command.baseCurrency(), command.quoteCurrency(), command.quantity(), command.price());
if (state.hasAkcesProcess(command.orderId())) {
OrderProcess sellOrderProcess = state.getAkcesProcess(command.orderId());
// cancel the reservation of the base currency
getCommandBus().send(new CancelReservationCommand(
state.userId(),
command.baseCurrency(),
command.orderId()));
// debit the base currency
log.info("CommandHandler: Debiting {} {} from user wallet", sellOrderProcess.amount(), command.baseCurrency());
getCommandBus().send(new DebitWalletCommand(
command.userId(),
command.baseCurrency(),
sellOrderProcess.size()));
// credit the quote currency (user receives quote currency for selling base currency)
BigDecimal quoteAmount = command.quantity().multiply(command.price());
log.info("CommandHandler: Crediting {} {} to user wallet", quoteAmount, command.quoteCurrency());
getCommandBus().send(new CreditWalletCommand(
command.userId(),
command.quoteCurrency(),
quoteAmount));
// TODO: we also need to update the counterparty wallet
return Stream.of(new SellOrderFilledEvent(
command.userId(),
command.orderId(),
command.counterpartyId(),
command.price(),
command.quantity(),
command.baseCurrency(),
command.quoteCurrency()
));
}
log.info("CommandHandler: No active process found for orderId={}", command.orderId());
return Stream.empty();
}

@EventHandler(produces = {BuyOrderPlacedEvent.class, SellOrderPlacedEvent.class}, errors = {})
public Stream<DomainEvent> handle(AmountReservedEvent event, OrderProcessManagerState state) {
log.info("EventHandler: Amount reserved for userId={}, currency={}, amount={}, referenceId={}",
event.userId(), event.currency(), event.amount(), event.referenceId());
Expand All @@ -195,22 +299,27 @@ public Stream<DomainEvent> handle(AmountReservedEvent event, OrderProcessManager
if (orderProcess != null) {
log.info("EventHandler: Placing market order for orderId={}, marketId={}",
orderProcess.orderId(), orderProcess.market().id());
Side side = orderProcess instanceof BuyOrderProcess ? Side.BUY : Side.SELL;
getCommandBus().send(new PlaceMarketOrderCommand(
orderProcess.market().id(),
orderProcess.orderId(),
state.userId(),
Side.BUY,
side,
orderProcess.amount(),
null));
return Stream.of(new BuyOrderPlacedEvent(state.userId(), orderProcess.orderId(), orderProcess.market(), orderProcess.amount(), null));
orderProcess.size()));
if (orderProcess instanceof BuyOrderProcess) {
return Stream.of(new BuyOrderPlacedEvent(state.userId(), orderProcess.orderId(), orderProcess.market(), orderProcess.amount(), null));
} else {
return Stream.of(new SellOrderPlacedEvent(state.userId(), orderProcess.orderId(), orderProcess.market(), orderProcess.size(), null));
}
} else {
log.info("EventHandler: No order process found for referenceId={}", event.referenceId());
// TODO: this cannot happen
return Stream.empty();
}
}

@EventHandler(produces = BuyOrderRejectedEvent.class, errors = {})
@EventHandler(produces = {BuyOrderRejectedEvent.class, SellOrderRejectedEvent.class}, errors = {})
public Stream<DomainEvent> handle(InsufficientFundsErrorEvent errorEvent, OrderProcessManagerState state) {
log.info("EventHandler: Insufficient funds error for userId={}, currency={}, available={}, requested={}, referenceId={}",
state.userId(), errorEvent.currency(), errorEvent.availableAmount(), errorEvent.requestedAmount(), errorEvent.referenceId());
Expand All @@ -222,7 +331,7 @@ public Stream<DomainEvent> handle(InsufficientFundsErrorEvent errorEvent, OrderP
}
}

@EventHandler(produces = BuyOrderRejectedEvent.class, errors = {})
@EventHandler(produces = {BuyOrderRejectedEvent.class, SellOrderRejectedEvent.class}, errors = {})
public Stream<DomainEvent> handle(InvalidCryptoCurrencyErrorEvent errorEvent, OrderProcessManagerState state) {
log.info("EventHandler: Invalid crypto currency error for userId={}, cryptoCurrency={}, referenceId={}",
state.userId(), errorEvent.cryptoCurrency(), errorEvent.referenceId());
Expand Down Expand Up @@ -259,6 +368,16 @@ public void handle(MarketOrderFilledEvent event, CommandBus commandBus) {
));
break;
case SELL:
commandBus.send(new FillSellOrderCommand(
event.ownerId(),
event.orderId(),
event.counterpartyId(),
event.price(),
event.quantity(),
event.baseCurrency(),
event.quoteCurrency()
));
break;
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
/*
* Copyright 2022 - 2025 The Original Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/

package org.elasticsoftware.cryptotrading.aggregates.orders;

import org.elasticsoftware.cryptotrading.aggregates.orders.commands.RejectOrderCommand;
import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket;
import org.elasticsoftware.cryptotrading.aggregates.orders.events.SellOrderRejectedEvent;
import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InsufficientFundsErrorEvent;
import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InvalidCryptoCurrencyErrorEvent;

import java.math.BigDecimal;

public record SellOrderProcess(
String orderId,
CryptoMarket market,
BigDecimal size,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

keep size and amount properties like in the BuyOrderProcess but set the amount to null in the constructor

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added amount field to SellOrderProcess record. Constructor now sets size to quantity and amount to null, matching BuyOrderProcess structure. Changes in commit 98bab9f.

BigDecimal amount,
String clientReference,
OrderProcessState state
) implements OrderProcess {
public SellOrderProcess(String orderId,
CryptoMarket market,
BigDecimal quantity,
String clientReference) {
this(orderId, market, quantity, null, clientReference, OrderProcessState.CREATED);
}

@Override
public String getProcessId() {
return orderId();
}

@Override
public SellOrderRejectedEvent handle(InsufficientFundsErrorEvent error) {
return new SellOrderRejectedEvent(error.walletId(), orderId(), clientReference());
}

@Override
public SellOrderRejectedEvent handle(InvalidCryptoCurrencyErrorEvent error) {
return new SellOrderRejectedEvent(error.walletId(), orderId(), clientReference());
}

@Override
public SellOrderRejectedEvent handle(RejectOrderCommand command) {
return new SellOrderRejectedEvent(command.userId(), orderId(), clientReference());
}

@Override
public OrderProcess withState(OrderProcessState state) {
return new SellOrderProcess(orderId(), market, size(), amount(), clientReference(), state);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
/*
* Copyright 2022 - 2025 The Original Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/

package org.elasticsoftware.cryptotrading.aggregates.orders.commands;

import jakarta.validation.constraints.NotNull;
import org.elasticsoftware.akces.annotations.AggregateIdentifier;
import org.elasticsoftware.akces.annotations.CommandInfo;
import org.elasticsoftware.akces.commands.Command;

import java.math.BigDecimal;

@CommandInfo(type = "FillSellOrder", version = 1)
public record FillSellOrderCommand(
@NotNull @AggregateIdentifier String userId,
@NotNull String orderId,
@NotNull String counterpartyId,
@NotNull BigDecimal price,
@NotNull BigDecimal quantity,
@NotNull String baseCurrency,
@NotNull String quoteCurrency
) implements Command {
@Override
public String getAggregateId() {
return userId();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/*
* Copyright 2022 - 2025 The Original Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/

package org.elasticsoftware.cryptotrading.aggregates.orders.commands;

import jakarta.validation.constraints.NotNull;
import org.elasticsoftware.akces.annotations.AggregateIdentifier;
import org.elasticsoftware.akces.annotations.CommandInfo;
import org.elasticsoftware.akces.commands.Command;
import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket;

import java.math.BigDecimal;

@CommandInfo(type = "PlaceSellOrder", version = 1)
public record PlaceSellOrderCommand(
@NotNull @AggregateIdentifier String userId,
@NotNull CryptoMarket market,
@NotNull BigDecimal quantity,
@NotNull String clientReference
) implements Command {
@Override
@NotNull
public String getAggregateId() {
return userId();
}
}
Loading
Loading