/*
    @Author: Bhaskar S
    @Blog:   https://www.polarsparc.com
    @Date:   04 Dec 2020
*/

syntax = "proto3";

package serverstream;

option go_package = "polarsparc.com/grpc/serverstream/currencypb";

option java_multiple_files = true;
option java_package = "com.polarsparc.gss";

message CurrencyRateRequest {
  string from = 1;
  string to = 2;
}

message CurrencyRateResponse {
  string agent = 1;
  string from = 2;
  string to = 3;
  double rate = 4;
}

service CurrencyService {
  rpc getCurrencyRate(CurrencyRateRequest) returns (stream CurrencyRateResponse) {};
}

.
.
.
type CurrencyServiceServer interface {
	GetCurrencyRate(*CurrencyRateRequest, CurrencyService_GetCurrencyRateServer) error
}
.
.
.

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package main

import (
  "fmt"
  "github.com/pkg/errors"
  "log"
  "strings"
)

type CurrencyRate struct {
  Agent string
  Rate float64
}

type RatesCache map[string][]CurrencyRate

type server struct {
  cache RatesCache
}

func (s *server) Init() {
  l1 := []CurrencyRate{{Agent: "Alice", Rate: 1.30}, {Agent: "Bob", Rate: 1.302},{Agent: "Dave", Rate: 1.31}}
  s.cache["USD:CAD"] = l1
  l2 := []CurrencyRate{{Agent: "Alice", Rate: 0.85}, {Agent: "Charlie", Rate: 0.84}}
  s.cache["USD:EUR"] = l2
  l3 := []CurrencyRate{{Agent: "Bob", Rate: 0.75}, {Agent: "Charlie", Rate: 0.751},{Agent: "Eve", Rate: 0.74}}
  s.cache["USD:GBP"] = l3
}

func (s *server) GetAgentRates(from string, to string) ([]CurrencyRate, error) {
  key := strings.ToUpper(from + ":" + to)

  log.Printf("Currency rate request for key: %s\n", key)

  rates := s.cache[key]
  if rates == nil {
    return nil, errors.New(fmt.Sprintf("No rate for currency from: %s, to: %s", from, to))
  }

  log.Printf("Currency rates for key: %s = %v", key, rates)

  return rates, nil
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package main

import (
  "google.golang.org/grpc"
  "log"
  "net"
  "polarsparc.com/grpc/serverstream/currencypb" // [1]
  "time"
)

func (s *server) GetCurrencyRate(req *currencypb.CurrencyRateRequest,
  stream currencypb.CurrencyService_GetCurrencyRateServer) error { // [2]
  log.Printf("Received a CurrencyRate request with req: %v\n", req)

  from := req.From
  to := req.To

  rates, err := s.GetAgentRates(from, to)
  if err == nil {
    log.Printf("Rates from agents: %v\n", rates)
    for _, r := range rates {
      res := ¤cypb.CurrencyRateResponse{Agent: r.Agent, From: from, To: to, Rate: r.Rate}
      stream.Send(res) // [3]
      time.Sleep(250 * time.Millisecond)
    }
    return nil
  }

  return err
}

func main()  {
  cs := &server{
    cache: RatesCache{},
  }
  cs.Init() // [4]

  log.Println("Ready to start the CurrencyRate server...")

  lis, err := net.Listen("tcp", "localhost:20002")
  if err != nil {
    log.Fatalf("Failed to create listener on localhost:20002")
  }

  srv := grpc.NewServer() // [5]

  currencypb.RegisterCurrencyServiceServer(srv, cs) // [6]

  if err = srv.Serve(lis); err != nil {
    log.Fatalf("Failed to start server: %v", err)
  }
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package main

import (
  "golang.org/x/net/context"
  "google.golang.org/grpc"
  "io"
  "log"
  "polarsparc.com/grpc/serverstream/currencypb"
)

func main() {
  log.Println("Ready to start the CurrencyRate client...")

  conn, err := grpc.Dial("localhost:20002", grpc.WithInsecure())
  if err != nil {
    log.Fatalf("Failed to connect to localhost:20002")
  }
  defer conn.Close()

  cl := currencypb.NewCurrencyServiceClient(conn) // [1]

  // Success
  req := ¤cypb.CurrencyRateRequest{From: "usd",
    To: "eur"} // [2]
  stream, err := cl.GetCurrencyRate(context.Background(), req) // [3]
  if err != nil {
    log.Fatalf("[1] Failed to send CurrencyRate request to localhost:20002")
  }
  for {
    res, err := stream.Recv() // [4]
    if err == io.EOF {
      break
    }
    if err != nil {
      log.Fatalf("[1] Received and error from CurrencyRate at localhost:20002: %v", err)
    }
    log.Printf("[1] ===> Agent: %s, Rate: %.03f\n", res.Agent, res.Rate)
  }

  // Error
  req2 := ¤cypb.CurrencyRateRequest{From: "usd",
    To: "jpy"}
  stream2, err := cl.GetCurrencyRate(context.Background(), req2)
  if err != nil {
    log.Fatalf("[2] Failed to send CurrencyRate request to localhost:20002")
  }
  for {
    res, err := stream2.Recv()
    if err == io.EOF {
      break
    }
    if err != nil {
      log.Fatalf("[2] Received and error from CurrencyRate at localhost:20002: %v", err)
    }
    log.Printf("[2] ===> Agent: %s, Rate: %.03f\n", res.Agent, res.Rate)
  }
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package com.polarsparc.gss.server;

public class CurrencyRate {
    private final String agent;
    private final Double rate;

    public CurrencyRate(String agent, Double rate) {
        this.agent = agent;
        this.rate = rate;
    }

    public String getAgent() {
        return this.agent;
    }

    public Double getRate() {
        return this.rate;
    }

    @Override
    public String toString() {
        return "CurrencyRate{" +
                "agent='" + agent + '\'' +
                ", rate=" + rate +
                '}';
    }
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package com.polarsparc.gss.server;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

public class CurrencyRateProvider {
    private final static Logger LOGGER = Logger.getLogger(CurrencyRateProvider.class.getName());

    private final static Map<String, List<CurrencyRate>>> ratesTable = new HashMap<>();

    static {
        LOGGER.setLevel(Level.INFO);

        ratesTable.put("USD:CAD", Arrays.asList(new CurrencyRate("Alice", 1.30),
                new CurrencyRate("Bob", 1.302),
                new CurrencyRate("Dave", 1.31)));
        ratesTable.put("USD:EUR", Arrays.asList(new CurrencyRate("Alice", 0.85),
                new CurrencyRate("Charlie", 0.84)));
        ratesTable.put("USD:GBP", Arrays.asList(new CurrencyRate("Bob", 0.75),
                new CurrencyRate("Charlie", 0.751),
                new CurrencyRate("Eve", 0.74)));
    }

    private CurrencyRateProvider() {
    }

    public static List<CurrencyRate>> getCurrencyRate(String from, String to) {
        String key = (from + ":" + to).toUpperCase();

        LOGGER.info(String.format("Currency rate request for key: %s", key));

        if (!ratesTable.containsKey(key)) {
            throw new RuntimeException(String.format("No rate for currency from: %s, to: %s", from, to));
        }

        List<CurrencyRate>> rates = ratesTable.get(key);

        LOGGER.info(String.format("Currency rates for key: %s = %s", key, rates));

        return rates;
    }
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package com.polarsparc.gss.server;

import com.polarsparc.gss.CurrencyRateRequest;
import com.polarsparc.gss.CurrencyRateResponse;
import com.polarsparc.gss.CurrencyServiceGrpc;

import io.grpc.Status;
import io.grpc.stub.StreamObserver;

import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

public class CurrencyRateService extends CurrencyServiceGrpc.CurrencyServiceImplBase { // [1]
    private final static Logger LOGGER = Logger.getLogger(CurrencyRateService.class.getName());

    static {
        LOGGER.setLevel(Level.INFO);
    }

    @Override
    public void getCurrencyRate(CurrencyRateRequest request, 
      StreamObserver<CurrencyRateResponse> responseObserver) { // [2]
        String from = request.getFrom();
        String to = request.getTo();

        List<CurrencyRate> rates;
        try {
            rates = CurrencyRateProvider.getCurrencyRate(from, to);
        }
        catch (RuntimeException ex) {
            Status status = Status.FAILED_PRECONDITION.withDescription(ex.getMessage());
            responseObserver.onError(status.asRuntimeException());
            return;
        }

        if (rates != null) {
            LOGGER.info(String.format("Rates from agents: %s", rates));

            rates.forEach(r -> {
                CurrencyRateResponse response = CurrencyRateResponse.newBuilder()
                        .setAgent(r.getAgent())
                        .setFrom(from)
                        .setTo(to)
                        .setRate(r.getRate())
                        .build();
                responseObserver.onNext(response); // [3]
                try {
                    Thread.sleep(250);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            responseObserver.onCompleted(); // [4]
        }
    }
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package com.polarsparc.gss.server;

import io.grpc.Server;
import io.grpc.ServerBuilder;

import java.io.IOException;

public class CurrencyRateServer {
    public static void main(String[] args) {
        Server server = ServerBuilder.forPort(20002) // [1]
                .addService(new CurrencyRateService()) // [2]
                .build();

        try {
            server.start(); // [3]
        } catch (IOException e) {
            e.printStackTrace();
        }

        System.out.print("Started the gRPC CurrencyRateService on 20002 ...\n");

        try {
            server.awaitTermination(); // [4]
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package com.polarsparc.gss.client;

import com.polarsparc.gss.CurrencyRateResponse;
import io.grpc.stub.StreamObserver;

import java.util.concurrent.CountDownLatch;

public class CurrencyRateStreamObserver implements StreamObserver<CurrencyRateResponse> {
    private final CountDownLatch latch;

    public CurrencyRateStreamObserver(CountDownLatch latch) {
        this.latch = latch;
    }

    @Override
    public void onNext(CurrencyRateResponse response) { // [1]
        System.out.printf("Agent: %s, Rate: %.03f\n", response.getAgent(), response.getRate());
    }

    @Override
    public void onError(Throwable ex) { // [2]
        System.out.println("Exception: " + ex.getMessage());
    }

    @Override
    public void onCompleted() { // [3]
        System.out.println("Done !!!");
        latch.countDown();
    }
}

/*
  @Author: Bhaskar S
  @Blog:   https://www.polarsparc.com
  @Date:   04 Dec 2020
*/

package com.polarsparc.gss.client;

import com.polarsparc.gss.CurrencyRateRequest;
import com.polarsparc.gss.CurrencyRateResponse;
import com.polarsparc.gss.CurrencyServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;

import java.util.Iterator;
import java.util.concurrent.CountDownLatch;

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class CurrencyRateClientTest {
    private CurrencyServiceGrpc.CurrencyServiceBlockingStub blockingStub;
    private CurrencyServiceGrpc.CurrencyServiceStub stub;

    @BeforeAll
    public void setup() {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 20002) // [1]
                .usePlaintext() // [2]
                .build();
        this.blockingStub = CurrencyServiceGrpc.newBlockingStub(channel); // [3]
        this.stub = CurrencyServiceGrpc.newStub(channel); // [4]
    }

    @Test
    public void currencyRateBlockingTestOne() {
        CurrencyRateRequest request = CurrencyRateRequest.newBuilder()
                .setFrom("usd")
                .setTo("eur")
                .build(); // [5]
        Iterator<CurrencyRateResponse> response = this.blockingStub.getCurrencyRate(request); // [6]
        response.forEachRemaining(res -> System.out.printf("Agent: %s, Rate: %.03f\n", res.getAgent(), res.getRate()));
    }

    @Test
    public void currencyRateBlockingTestTwo() {
        CurrencyRateRequest request = CurrencyRateRequest.newBuilder()
                .setFrom("eur")
                .setTo("jpy")
                .build(); // [5]
        Iterator<CurrencyRateResponse> response = this.blockingStub.getCurrencyRate(request); // [6]
        Assertions.assertThrows(io.grpc.StatusRuntimeException.class, () -> response.forEachRemaining(res ->
                System.out.printf("Agent: %s, Rate: %.03f\n", res.getAgent(), res.getRate())));
    }

    @Test
    public void currencyRateAsyncTestOne() {
        CountDownLatch latch = new CountDownLatch(1);
        CurrencyRateRequest request = CurrencyRateRequest.newBuilder()
                .setFrom("usd")
                .setTo("cad")
                .build(); // [5]
        this.stub.getCurrencyRate(request, new CurrencyRateStreamObserver(latch)); // [7]
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}