Spring Cloud CommandRouter

[smcvb]

An implementation of the CommandRouter based on Spring-Cloud-Commons.
When a Spring-Cloud compatible server is present and additional applications annotated with @EnableDiscoveryClient are fired up, ServiceInstance will be created.
The SpringCloudCommandRouter updates the LocalServiceInstance it's Metadata with the load factor and command filter. The SpringCloudCommandRouter has an @eventlistener annotated method to listen to every HearthbeatEvent and will check whether it's set of Axon-nodes has to be updated and will if necessary.

Additionally a SpringHttpCommandBusConnector has been addded for command sending/receiving based on the RestTemplate and RestController.

[sganslandt]

Interesting stuff. We've built something very similar both for Axon 2 and 3, and it would definitely be beneficial to something like this being part of the core offering. Why was the PR closed?

[smcvb]

No worries @sganslandt, has been opened up again!
I was trying out some things.

[jorgheymans]

@smcvb IIUC this is an alternative to jgroups based command routing, based on for example zookeeper instead ?

[smcvb]

@jorgheymans yes it is. For testing this stuff I've created an Axon application through Spring Boot which has Netflix's Eureka Client on the pom, together with a Eureka Server running to regulate it all

[nickdk]

I know I've asked this in other places already (axon mailing list) but would be neat if we could have some example project or some documentation. I just tried to get an example working myself with a Eureka server and a few clients. Everything seems to work except that I can never get the updateMemberships to trigger when new clients register themselves, causing the whole consistentHash to only refer to one member. I'm new to the whole Eureka thing so I might be doing something wrong there, I'm gonna keep trying to get it working but again it would be lovely if someone has some extra info or a working example :-)

[smcvb]

@nickdk I'm actually planning to send out a example project for this, maybe through a blog, but cannot give you a ETA for that right now.

How I tested it, was by running a Spring Boot Eureka Server and running several Spring Boot applications which where Eureka clients and Axon nodes. What did you do Axon configuration-wise?

[alexey-krylov]

@nickdk <https://github.com/nickdk>, maybe this config sample can help: @configuration @ConditionalOnProperty("axon.distributed.command-bus.enabled") @Profile('!' + ApplicationProfiles.AUTO) public class DistributedCommandBusConfig { @qualifier("localSegment") @bean public CommandBus localSegment(TransactionManager transactionManager) { SimpleCommandBus localSegment = new SimpleCommandBus(transactionManager, NoOpMessageMonitor.INSTANCE); localSegment.registerDispatchInterceptor(new BeanValidationInterceptor<>()); localSegment.registerHandlerInterceptor(new LoggingInterceptor<>()); return localSegment; } @bean public SpringHttpCommandBusConnector connector( @qualifier("localSegment") CommandBus commandBus, RestTemplate restRemplate, Serializer serializer ) { return new SpringHttpCommandBusConnector(commandBus, restRemplate, serializer); } @bean @primary public DistributedCommandBus distributedCommandBus( DiscoveryClient discoveryClient, Serializer serializer, SpringHttpCommandBusConnector connector ) { return new DistributedCommandBus(new SpringCloudCommandRouter( discoveryClient, new AnnotationRoutingStrategy(), serializer ), connector); } } 2017-01-17 19:31 GMT+03:00 Steven <notifications@github.com>:

…

@nickdk <https://github.com/nickdk> I'm actually planning to send out a example project for this, maybe through a blog, but cannot give you a ETA for that right now. How I tested it, was by running a Spring Boot Eureka Server and running several Spring Boot applications which where Eureka clients and Axon nodes. What did you do Axon configuration-wise? — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#214 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AA6N8Ck5XXYyP3zPp9Bfpza_-VEaUtWvks5rTOzxgaJpZM4K8TWp> .

-- С уважением, Алексей Крылов mailto: Aleksey.Krylov@gmail.com

[nickdk]

Thanks already guys. I was able to get some progress, the @eventlistener on the SpringCloudCommandRouter wasn't getting picked up by Spring scanning, fixed that. Now I'm in the situation where I have one ConfigServer running (that also serves as an AxonClient) and two separate AxonClient instances. All running localhost on different ports (server 8761 and clients 8762 and 8763). These two classes:

@EnableEurekaServer
@SpringBootApplication(scanBasePackages = "org.axonframework")
public class ConfigServer {

	@Autowired
	private DiscoveryClient discoveryClient;

	public static void main(String[] args) throws InterruptedException {
		SpringApplication.run(ConfigServer.class, args);
	}

	@Bean
	public CommandGateway commandGateway() {
		return new DefaultCommandGateway(commandBus());
	}

	@Bean(initMethod = "start")
	public SubscribingEventProcessor eventProcessor(){
		EventStore eventStore = new EmbeddedEventStore(new InMemoryEventStorageEngine());
		Repository<Item> repository = new EventSourcingRepository<>(Item.class, eventStore);

		new AggregateAnnotationCommandHandler<>(Item.class, repository).subscribe(commandBus());

		return new SubscribingEventProcessor("processor", new SimpleEventHandlerInvoker((EventListener) event -> {
			ItemCreated itemCreated = (ItemCreated) event.getPayload();
			System.out.println("Received an ItemCreated event with id: " + itemCreated.getItemId());
		}), eventStore);
	}

	@Bean
	public SpringHttpCommandBusConnector springHttpCommandBusConnector() {
		CommandBus commandBus = new SimpleCommandBus();

		RestTemplate restTemplate = new RestTemplate();
		return new SpringHttpCommandBusConnector(commandBus, restTemplate, new XStreamSerializer());
	}

	@Bean
	public SpringCloudCommandRouter commandRouter() {
		return new SpringCloudCommandRouter(discoveryClient, new AnnotationRoutingStrategy(), new XStreamSerializer());
	}

	@Bean
	public CommandBus commandBus() {
		return new DistributedCommandBus(commandRouter(), springHttpCommandBusConnector());
	}

}

@EnableDiscoveryClient
@SpringBootApplication(scanBasePackages = "org.axonframework")
public class AxonClient {

	@Autowired
	private DiscoveryClient discoveryClient;

	public static void main(String[] args) throws InterruptedException {
		SpringApplication.run(AxonClient.class, args);
	}

	@Bean
	public CommandGateway commandGateway() {
		return new DefaultCommandGateway(commandBus());
	}

	@Bean(initMethod = "start")
	public SubscribingEventProcessor eventProcessor(){
		EventStore eventStore = new EmbeddedEventStore(new InMemoryEventStorageEngine());
		Repository<Item> repository = new EventSourcingRepository<>(Item.class, eventStore);

		new AggregateAnnotationCommandHandler<>(Item.class, repository).subscribe(commandBus());

		return new SubscribingEventProcessor("processor", new SimpleEventHandlerInvoker((EventListener) event -> {
			ItemCreated itemCreated = (ItemCreated) event.getPayload();
			System.out.println("Received an ItemCreated event with id: " + itemCreated.getItemId());
		}), eventStore);
	}

	@Bean
	public SpringHttpCommandBusConnector springHttpCommandBusConnector() {
		CommandBus commandBus = new SimpleCommandBus();

		RestTemplate restTemplate = new RestTemplate();
		return new SpringHttpCommandBusConnector(commandBus, restTemplate, new XStreamSerializer());
	}

	@Bean
	public CommandBus commandBus() {
		SpringCloudCommandRouter commandRouter = new SpringCloudCommandRouter(discoveryClient, new AnnotationRoutingStrategy(), new XStreamSerializer());

		return new DistributedCommandBus(commandRouter, springHttpCommandBusConnector());
	}
}

I also have a simple RestController -> commandGateway that I can use to push some commands through a HTTP endpoint. When I push commands the commands are always executed on the instance I target with the port.

When I debug I see that the atomicConsistentHash of the router only contains one member, I then tried debugging the updateMemberships I noticed something odd:

You can see that the code evaluation sees 3 instances (these are the correct 8761, 8762 and 8763 instances). Yet in the breakpoint I can see that only 1 of them is passed to the private updateMemberships that actually adds the instances to the atomicConsistentHash. So seems like the 3 instances get reduced to 1 by the stream() chain.

Maybe I'm doing something wrong configuration wise? Anyway, gonna see if I can progress tomorrow, if not I'll have a look at your example @alexey-krylov, thanks!

[smcvb]

The only diff I currently can point out, compared to your set up, is that my Eureka server was really just a Eureka server, nothing more, whilst yours is server and client in one.

Have you tried debugging in the ConsistentHash class? There is a call on the ConsistentHash.with(...) function that check whether the member is already contained in the set. If so, it just returns the ConsistentHash as is instead of adding a new member. Although your debug screen cap shows three instances, it does not tell me their contents so I'm not sure if they are equal when they're converted to a ConsistentHashMember.

[nickdk]

Yeah it indeed seems that for some reason they are seen as one member, but they already get reduced from 3 to 1 even before the ConsistentHash gets into play. Anyway, I hopefully should have some free hours this afternoon, I'll keep you posted ;-)