package io.cax.forex.ingester.services;

import io.cax.forex.ingester.Utils;
import io.cax.forex.ingester.domain.Tick;
import io.cax.forex.ingester.repositories.TickRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.ClientHttpRequest;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestOperations;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponents;
import org.springframework.web.util.UriComponentsBuilder;

import java.util.Scanner;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by cq on 17/4/16.
 */
@Service
@EnableAsync
public class TickService {

    Logger logger = LoggerFactory.getLogger(TickService.class);

    @Value("${oanda.access_token}")
    private String accessToken;

    @Value("${oanda.account_id}")
    private String accountId;


    @Value("${oanda.domain}")
    private String domain;


    private RestOperations restTemplate;

    private TickRepository repository;

    @Autowired
    public void setRepository(TickRepository repository) {
        this.repository = repository;
    }

    private InstrumentsService instrumentsService;

    @Autowired
    public void setInstrumentsService(InstrumentsService instrumentsService) {
        this.instrumentsService = instrumentsService;
    }

    private AtomicBoolean running = new AtomicBoolean();


    /**
     * Starts the data streaming.
     */
    @Async
    public void startStreaming(){

        if(running.get()==true) return; //already running.

        running.compareAndSet(false,true);

        restTemplate = new RestTemplate();

        UriComponents uriComponents = UriComponentsBuilder
                .fromUriString(domain)
                .path("/v1/prices")
                .queryParam("accountId",accountId)
                .queryParam("instruments",instrumentsService.instruments()).build();

        restTemplate.execute(uriComponents.toUriString(),
                HttpMethod.GET,
                clientHttpRequest -> setHeaders(clientHttpRequest),
                clientHttpResponse -> {
                    try(Scanner scanner = new Scanner(clientHttpResponse.getBody(),"utf-8")){

                        while(scanner.hasNext() && running.get()) saveTick(scanner.nextLine());
                    }
                    return new ResponseEntity<>(HttpStatus.OK);
                });
    }

    /**
     * Stops the data streaming
     */
    public void stopStreaming(){

        running.compareAndSet(true, false);
    }


    /**
     * Processes the data and saves the tick if pertinent. Hearbeat ticks are not saved.
     * @param content
     */
    private void saveTick(String content){

        try{

            Tick tick = Utils.convertToTick(content);
            if(tick!=null) repository.save(tick);

        } catch(Exception e){
            logger.error("Error saving tick: " + e.getMessage());
        }


    }


    private ClientHttpRequest setHeaders(ClientHttpRequest clientHttpRequest){
        clientHttpRequest.getHeaders().add("X-Accept-Datetime-Format","UNIX");
        clientHttpRequest.getHeaders().add("Authorization","Bearer " + accessToken);
        return clientHttpRequest;
    }

    public long ticksProcessed(){
        logger.debug("ticksProcessed called!!");
        return repository.count();
    }

}

@@ -1359,16 +1359,22 @@
 private 
+final 
 RestOper
@@ -1392,16 +1392,37 @@
 Template
+ = new RestTemplate()
 ;%0A%0A    p
@@ -2045,52 +2045,8 @@
 );%0A%0A
-        restTemplate = new RestTemplate();%0A%0A
     
@@ -2308,24 +2308,42 @@
 ).build();%0A%0A
+        try%7B%0A%0A    
         rest
@@ -2404,16 +2404,20 @@
         
+   
+ 
 HttpMeth
@@ -2424,16 +2424,20 @@
 od.GET,%0A
+    
         
@@ -2504,32 +2504,36 @@
                 
+    
 clientHttpRespon
@@ -2560,16 +2560,20 @@
         
+ 
+   
 try(Scan
@@ -2663,16 +2663,20 @@
         
+    
 while(sc
@@ -2759,18 +2759,26 @@
         
+   
+ 
 %7D%0A
+    
         
@@ -2853,11 +2853,128 @@
     
-%7D);
+    %7D);%0A%0A        %7Dcatch(Exception e)%7B%0A            logger.error(%22Error reading from the remote connection%22,e);%0A        %7D%0A
 %0A   

Add try/catch to catch exception on http conn.