HystrixWrapper.java

import java.util.Optional;
import java.util.Set;

import no.finntech.search.front.Group;
import no.finntech.search.front.Location;

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public final class HystrixWrapper extends HystrixCommand<Group> {
    private static final Logger LOG = LogManager.getLogger();
    private final Source source;
    private final String query;
    private final Optional<Location> location;
    private final Set<String> searchKeys;
    private final Optional<String> cookieToken;


    public HystrixWrapper(Source source,
                          String query,
                          Optional<Location> location,
                          Set<String> searchKeys,
                          Optional<String> cookieToken) {
        super(Setter
                        .withGroupKey(HystrixCommandGroupKey.Factory.asKey("no.finntech.search.front.sources"))
                        .andCommandKey(HystrixCommandKey.Factory.asKey(source.toString()))
        );

        this.source = source;
        this.query = query;
        this.location = location;
        this.searchKeys = searchKeys;
        this.cookieToken = cookieToken;
    }

    @Override
    protected Group run() throws Exception {
        Group suggest = source.suggest(query, location, searchKeys, cookieToken);
        LOG.trace("Getting suggestions from " + source +
                ". query: " + query +
                ", loc: " + location +
                ", searchkey: " + searchKeys +
                ", num suggestions: " + suggest.getCount());

        return suggest;
    }

    @Override
    protected Group getFallback() {
        int timeUsed = getExecutionTimeInMilliseconds();
        LOG.info(source + " returning fallback. time used: " + timeUsed);
        return source.getFallback(timeUsed);
    }
}

SseEmitter.java

@RequestMapping("/asyncsuggest")
    public ResponseEntity<SseEmitter> asyncSuggest(@RequestParam(value = "term", required = false) String term,
                                                   @RequestParam(value = "lat", required = false) Optional<Double> lat,
                                                   @RequestParam(value = "lon", required = false) Optional<Double> lon,
                                                   @RequestParam(value = "searchKeys", required = false) String searchKeys,
                                                   @RequestParam(value = "client", defaultValue = "MFINN", required = false) FrontClient client,
                                                   @RequestParam(value = "includeEmpty", defaultValue = "false", required = false) boolean includeEmpty,
                                                   HttpServletRequest request, HttpServletResponse response) {
        Optional<Location> location = getLocation(lat, lon);
        Set<String> searchKeySet = getSearchKeys(searchKeys);

        SseEmitter emitter = new SseEmitter();
        StopWatch sw = new StopWatch("Frontcontroller ASYNC");
        sw.start("Build Groups");
        final Optional<String> cookieToken = CookieTokenMapper.getCookieToken(request);
        final Observable<Group> allGroups = Observable
                .merge(sources.stream()
                              .map(s -> new HystrixWrapper(s, term, location, searchKeySet, cookieToken).toObservable())
                              .collect(toList()))
                .filter(g -> includeEmpty || g.hasResults())
                .map(g -> cleanResults(g, client))
                .subscribeOn(Schedulers.io());
        allGroups.subscribe((group) -> {
                                try {
                                    emitter.send(event().data(group));
                                } catch (IOException ex) {
                                    LOG.warn("Failed to send event for group: {}", group, ex);
                                }
                            },
                            thr -> handleComplete(emitter, sw, thr),
                            () -> handleComplete(emitter, sw, null));

        return new ResponseEntity<>(emitter, HttpStatus.OK);
    }