Rick

Rick
Rick

Thursday, July 23, 2015

QBit: Intercepting method calls, grabbing http request, using BeforeMethodCall, AOP like features with qbit

Recently someone asked me if you could capture the request parameters from a request with QBit REST support. You can.
QBit has this interface.

BeforeMethodCall

package io.advantageous.qbit.service;

import io.advantageous.qbit.message.MethodCall;

/**
 * Use this to register for before method calls for services.
 * <p>
 * created by Richard on 8/26/14.
 *
 * @author rhightower
 */
public interface BeforeMethodCall {

    boolean before(MethodCall call);
}
With this BeforeMethodCall interface you can intercept a method call. If you register it with a ServiceQueue via the ServiceBuilder then the method interception happens in the same thread as the service queue method calls.
If you return false from the before method then the call will not be made. You can also intercept calls at the ServiceBundle and the ServiceEndpointServer levels using theEndpointServerBuilder and the ServiceBundleBuilder. When you register aBeforeMethodCall with a service bundle or and server end point, it gets called before the method is enqueued to the actual service queue. When you register aBeforeMethodCall with a service queue, it gets called right before the method gets invoked in the same thread as the service queue, i.e., in the service thread, which is most useful for capturing the HttpRequest.
But let's say that you want to access the HttpRequest object to do something special with it. Perhaps read the request params.
This is possible. One merely has to intercept the call. Every Request object has a property called originatingRequest. A MethodCall is a Request object as is anHttpRequest. This means that you just have to intercept the call withBeforeMethodCall grab the methodCall, and then use it to get the HttpRequest.

Service example

/**
 * Created by rhightower.
 */
@RequestMapping("/api")
public class PushService {

    private final ThreadLocal<HttpRequest> currentRequest;

    public PushService() {
        this.currentRequest = new ThreadLocal<>();
    }


    @RequestMapping(value = "/event", method = RequestMethod.POST)
    public void event(final Callback<Boolean> callback, final Event event) {

        final HttpRequest httpRequest = currentRequest.get();
        System.out.println(httpRequest.address());
        System.out.println(httpRequest.params().size());
        ...
    }
Now in the main method, we will need to construct the service and then register the service with the endpoint.
Notice the private final ThreadLocal<HttpRequest> currentRequest; because we will use that to store the current http request.

Register a ServiceQueue with an end point server

final ManagedServiceBuilder managedServiceBuilder = ManagedServiceBuilder.managedServiceBuilder();

        final PushService pushService = new PushService();


        final ServiceEndpointServer serviceEndpointServer =  
                     managedServiceBuilder.getEndpointServerBuilder()
                                                            .setUri("/")
                                                            .build();

      ...


        final ServiceQueue pushServiceQueue = ...
        serviceEndpointServer.addServiceQueue("/api/pushservice", pushServiceQueue);
Notice when we create the service queue separately we have to register the address it is bound under.
When we create the service queue (pushServiceQueue) for pushService, we want to tell it to use the same response queue as our endpoint server and register the beforeCall lambda to capture the HttpRequest from the MethodCall.

Creating a lambda expression to populate the currentRequest from the originatingRequest of the MethodCall (call)

        final ServiceQueue pushServiceQueue = managedServiceBuilder
                .createServiceBuilderForServiceObject(pushService) 
                .setResponseQueue(serviceEndpointServer.serviceBundle().responses())
                .setBeforeMethodCall(call -> {

                    pushService.currentRequest.set((HttpRequest) call.originatingRequest());
                    return true;
                })
                .buildAndStart();
The full example is a bit longer as it has some other things not mentioned in this article.
public class Event {

    private final String name;

    public Event(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

}

....


import io.advantageous.qbit.annotation.*;
import io.advantageous.qbit.admin.ManagedServiceBuilder;
import io.advantageous.qbit.http.request.HttpRequest;
import io.advantageous.qbit.reactive.Callback;
import io.advantageous.qbit.reactive.Reactor;
import io.advantageous.qbit.reactive.ReactorBuilder;
import io.advantageous.qbit.server.ServiceEndpointServer;
import io.advantageous.qbit.service.ServiceQueue;

import java.util.concurrent.TimeUnit;



/**
 * Created by rhightower.
 */
@RequestMapping("/api")
public class PushService {


    private final Reactor reactor;
    private final StoreServiceClient storeServiceClient;

    private final ThreadLocal<HttpRequest> currentRequest;

    public PushService(final Reactor reactor,
                       final StoreServiceClient storeServiceClient) {
        this.reactor = reactor;
        this.storeServiceClient = storeServiceClient;
        this.currentRequest = new ThreadLocal<>();
    }

    @RequestMapping("/hi")
    public String sayHi() {
        return "hi";
    }

    @RequestMapping(value = "/event", method = RequestMethod.POST)
    public void event(final Callback<Boolean> callback, final Event event) {

        final HttpRequest httpRequest = currentRequest.get();

        System.out.println(httpRequest.address());

        System.out.println(httpRequest.params().baseMap());
        storeServiceClient.addEvent(callback, event);

    }

    @QueueCallback({QueueCallbackType.LIMIT, QueueCallbackType.EMPTY, QueueCallbackType.IDLE})
    public void load() {

        reactor.process();
    }


    public static void main(String... args) {


        /* Using new snapshot 2. */
        final ManagedServiceBuilder managedServiceBuilder = ManagedServiceBuilder.managedServiceBuilder();

        final StoreService storeService = new StoreService();


        final ServiceQueue serviceQueue = managedServiceBuilder.createServiceBuilderForServiceObject(storeService)
                .buildAndStartAll();

        final StoreServiceClient storeServiceClient = serviceQueue.createProxyWithAutoFlush(StoreServiceClient.class,
                100, TimeUnit.MILLISECONDS);


        final PushService pushService = new PushService(ReactorBuilder.reactorBuilder().build(),
                storeServiceClient);

        final ServiceEndpointServer serviceEndpointServer = managedServiceBuilder.getEndpointServerBuilder()
                .setUri("/")
                .build();

        final ServiceQueue pushServiceQueue = managedServiceBuilder
                .createServiceBuilderForServiceObject(pushService)
                .setResponseQueue(serviceEndpointServer.serviceBundle().responses())
                .setBeforeMethodCall(call -> {

                    pushService.currentRequest.set((HttpRequest) call.originatingRequest());
                    return true;
                })
                .buildAndStart();

        serviceEndpointServer.addServiceQueue("/api/pushservice", pushServiceQueue);


        serviceEndpointServer.startServer();

        /* Wait for the service to shutdown. */
        managedServiceBuilder.getSystemManager().waitForShutdown();

    }

}
...

public class StoreService {

    public boolean addEvent(final Event event) {

        return true;
    }

}
...

import io.advantageous.qbit.reactive.Callback;

public interface StoreServiceClient {

    void addEvent(final Callback<Boolean> callback, final Event event);
}
...
import io.advantageous.boon.json.JsonFactory;
import io.advantageous.qbit.http.HTTP;

import static io.advantageous.boon.core.IO.puts;

public class TestMain {

    public static void main(final String... args) throws Exception {


        HTTP.Response hello = HTTP.jsonRestCallViaPOST("http://localhost:9090/api/event", JsonFactory.toJson(new Event("hello")));

        puts(hello.body(), hello.status());
    }
}
Kafka and Cassandra support, training for AWS EC2 Cassandra 3.0 Training