https://github.com/scalecube/scalecube-services ⚛ High-Speed ⚛ Cloud-Native ⚛ Reactive Microservices ⚓
@Configuration
@AutoConfigureAfter(CodecConfiguration::class)
@ConditionalOnClass(RSocketServiceTransport::class)
@ConditionalOnBean(ServiceMessageCodec::class)
class RsocketTransportConfiguration {
@Bean
fun rsocketServiceTransport(scalecubeProperties: ScalecubeProperties,
headersCodec: HeadersCodec, dataCodecs: List<DataCodec>): ServiceTransport {
var serviceTransport = RSocketServiceTransport()
scalecubeProperties.cluster?.transport?.let { transport ->
serviceTransport = serviceTransport.tcpServer {
TcpServer.create().runOn(it).port(transport.port)
}
transport.rsocket?.let { rsocket ->
serviceTransport = serviceTransport.numOfWorkers(rsocket.workerCount)
}
}
serviceTransport = serviceTransport.dataCodecs(dataCodecs)
serviceTransport = serviceTransport.headersCodec(headersCodec)
return serviceTransport
}
}package be.moesmedia.imera.projects.gateway.configuration.codecs
import be.moesmedia.imera.projects.gateway.configuration.SpringContext
import com.fasterxml.jackson.databind.ObjectMapper
import io.scalecube.services.transport.api.DataCodec
import io.scalecube.services.transport.api.HeadersCodec
import java.io.InputStream
import java.io.OutputStream
import java.lang.reflect.Type
class CustomJacksonCodec : DataCodec, HeadersCodec {
companion object {
const val contentType = "application/json"
}
private lateinit var objectMapper: ObjectMapper
override fun contentType(): String = contentType
override fun encode(stream: OutputStream?, headers: MutableMap<String, String>?) = getObjectMapper().writeValue(stream, headers)
@Suppress("UNCHECKED_CAST")
override fun decode(stream: InputStream): MutableMap<String, String> = when (stream.available()) {
0 -> HashMap()
else -> (getObjectMapper().readValue(stream, HashMap::class.java)) as HashMap<String, String>
}
override fun encode(stream: OutputStream, value: Any) =
getObjectMapper().writeValue(stream, value)
override fun decode(stream: InputStream, type: Type): Any = getObjectMapper().readValue(stream, getObjectMapper().typeFactory.constructType(type))
private fun getObjectMapper() = if (!this::objectMapper.isInitialized) {
this.objectMapper = SpringContext.getBean(ObjectMapper::class.java)!!
this.objectMapper
} else {
this.objectMapper
}
}
SpringContext is a simple Component that implements the ApplicationContextAware interface, which holds a reference towards the springcontext from which I can load the correct objectmapper
@Component
class SpringContext : ApplicationContextAware {
@Throws(BeansException::class)
override fun setApplicationContext(context: ApplicationContext) {
Companion.context = context
}
companion object {
private var context: ApplicationContext? = null
fun <T : Any?> getBean(beanClass: Class<T>): T? {
return context?.getBean(beanClass)
}
}
}
I learned a lot this weekend how to configure Scalecube in a Spring application, what are the development guidelines?
Hi @ronenhamias , is there an example somewhere on ScaleCube working with an Aeron transport? I was also curious to see ScaleCube working with something like an SBE codec (or any codec based on buffer-wrapping rather than field copying). When I looked at SBE with RSocket it seemed inefficient due to buffer-copying between the Netty ByteBufs and the Agrona buffers.
We do it internally in our enterprise project we didnt open that to the open-source
as part of our project
http://exberry.io/
but there is a project called Aeron POC where we did all our research and reserach it self is public info
@felipevillar
there is another project
so reactor team moved reactor aeron to our hands
"When I looked at SBE with RSocket it seemed inefficient due to buffer-copying between the Netty ByteBufs and the Agrona buffers"
you are very much correct there is an issue with reactor and netty.
you are very much correct there is an issue with reactor and netty.
so we have our own transport since in scalecube we are not bounded to one transport / protocol / codecs
and still Reactor-Aeron gives better perfromance then Reactor-netty (you can check the benchmarks)
but its not a replacement to reactor netty since these should be used on diffrent usecases
I see thanks @ronenhamias , that sounds interesting. Since the buffers underlying SBE codecs are mutable and shared (and typically accessed from a single thread) I was mainly interested in seeing how well such a codec would fit with the "share nothing" & reactive style of programming. I'll have a closer look at the scalecube/reactor-aeron project, maybe that will give me a better understanding.
and we internally have reactor cluster for our high frequency trading platform running production
just checking... by "reactor cluster" did you mean a ScaleCube cluster, or a real-logic/aeron-cluster?
scalecube provides 2 phase desrialization its very much desired that data will be mutable
@felipevillar
and yes for that transport we use single thread approch
its not a problem since on such critical path where it brings value it handles millions of messages per sec with low latnecy
ofcourse you must be careful not to block
also with Aeron you cant block so it fits very well
Hi, I met an error that I don't understand, would someone please take a look?
Thanks.
The error is:
Error receiving frame:
java.lang.IndexOutOfBoundsException: readerIndex(0) + length(4) exceeds writerIndex(1): AbstractPooledDerivedByteBuf$PooledNonRetainedSlicedByteBuf(ridx: 0, widx: 1, cap: 1/1, unwrapped: PooledUnsafeDirectByteBuf(ridx: 4, widx: 427, cap: 1024))I am using:
- <spring-boot.version>2.2.6.RELEASE</spring-boot.version>
- <scalecube.version>2.9.0</scalecube.version>
- <scalecube-cluster.version>2.4.10</scalecube-cluster.version>
- <rsocket.version>1.0.0-RC6</rsocket.version>
Code to reproduce:
@Slf4j
public class HelloScalecubeApplicationTest {
public interface Hello {
Mono<Void> hello();
}
public class HelloImpl implements Hello {
@Override
public Mono<Void> hello() {
log.info("hello");
return Mono.empty();
}
}
@Test
public void testScalecubeServices() {
Hello hello = mock(Hello.class);
when(hello.hello()).thenReturn(Mono.empty());
Microservices seed = Microservices.builder()
.discovery(ScalecubeServiceDiscovery::new)
.transport(RSocketServiceTransport::new)
.startAwait();
Microservices proxy = Microservices.builder()
.discovery(endpoint -> new ScalecubeServiceDiscovery(endpoint)
.membership(membership -> membership.seedMembers(seed.serviceAddress())))
.transport(RSocketServiceTransport::new)
.startAwait();
Microservices stub = Microservices.builder()
.discovery(endpoint -> new ScalecubeServiceDiscovery(endpoint)
.membership(membership -> membership.seedMembers(seed.serviceAddress())))
.transport(RSocketServiceTransport::new)
.services(new HelloImpl())
.startAwait();
proxy.call().api(Hello.class).hello().block();
verify(hello, times(1)).hello();
}
}
Oops, there's an mistake in the code snippet, I updated it and the error remains.
@Slf4j
public class HelloScalecubeApplicationTest {
public interface Hello {
Mono<Void> hello();
}
@Test
public void testScalecubeServices() {
Hello hello = mock(Hello.class);
when(hello.hello()).thenReturn(Mono.empty());
Microservices seed = Microservices.builder()
.discovery(ScalecubeServiceDiscovery::new)
.transport(RSocketServiceTransport::new)
.startAwait();
Microservices proxy = Microservices.builder()
.discovery(endpoint -> new ScalecubeServiceDiscovery(endpoint)
.membership(membership -> membership.seedMembers(seed.serviceAddress())))
.transport(RSocketServiceTransport::new)
.startAwait();
Microservices stub = Microservices.builder()
.discovery(endpoint -> new ScalecubeServiceDiscovery(endpoint)
.membership(membership -> membership.seedMembers(seed.serviceAddress())))
.transport(RSocketServiceTransport::new)
.services(hello)
.startAwait();
proxy.call().api(Hello.class).hello().block();
verify(hello, times(1)).hello();
}
}
if you scroll up you can see a config I made for running scalecube with spring (sidenote: this is from a POC, I don't necessarily use best practices)
the biggest caveat you have to take into account is how Scalecube instantiates it's codecs
Scalecube utilizes the ServiceLoader which uses the non-args constructor to create a new instance of it, however by doing this Scalecube instantiates it outside of the spring-context so you don't have any reference to your spring beans
that's why I created the SpringContext class which implements the ApplicationContextAware interface, when the applicationContext is initialized this class will get a reference to it and stores it in a static reference that can be used in my CustomJacksonCodec
I fixed the annotations problem, and I have removed spring from the hello world program.
The error still persist.
Then, I removed the seed, run scalecube-seed and use it as seed, the error is gone.
So, I think it is related to the scalecube-cluster (since it is happens while connecting to seed), but I still can't figure out why.
The error still persist.
Then, I removed the seed, run scalecube-seed and use it as seed, the error is gone.
So, I think it is related to the scalecube-cluster (since it is happens while connecting to seed), but I still can't figure out why.
the question is what the benefit of breaking it down?
the idea is that it provides a single entry point to all the sub modules
i guess its matter of style but i guess when you break something it should bring also new value
like some reuse... or some flexibility...
in such case better to point on some specific value we can gain.