// Path: grpc-mate-java/elasticsearch-service/src/main/java/io/datanerd/es/dao/ProductDao.java
// @Singleton
// public class ProductDao {
// 
//   private static Logger log = LoggerFactory.getLogger(ProductDao.class); //NOPMD
//   public static final String INDEX = "products";
//   public static final String TYPE = "Product";
//   public static final TimeValue DEFAULT_SCROLL_TIME_VALUE = new TimeValue(60000);
//   static final int SCROLL_SIZE = 50;
// 
//   @Inject
//   private TransportClient esClient;
// 
//   @Inject
//   private JsonFormat.Printer jsonPrinter;
// 
//   @Inject
//   private JsonFormat.Parser jsonParser;
// 
//   /**
//    * Upsert the given product into ES index.
//    *
//    * @param product given product object
//    */
//   public void upsertProduct(Product product) throws InvalidProtocolBufferException {
//     log.debug("save product into ES");
//     final UpdateRequestBuilder updateRequestBuilder =
//         esClient
//             .prepareUpdate(INDEX, TYPE, String.valueOf(product.getProductId()))
//             .setDoc(
//                 jsonPrinter
//                     .includingDefaultValueFields()
//                     .omittingInsignificantWhitespace()
//                     .print(product)
//             )
//             .setDocAsUpsert(true);
//     updateRequestBuilder.get();
//   }
// 
//   /**
//    * Init index setting and mapping.
//    *
//    * @return true if a new index was created, false otherwise
//    */
//   public boolean initIndexIfNotExists() throws IOException {
//     final IndicesExistsResponse existsResponse = esClient.admin().indices().prepareExists(INDEX).get();
//     if (existsResponse.isExists()) {
//       return false;
//     }
//     final String settings = Resources.toString(
//         getClass().getResource("/elasticsearch/product_settings.json"),
//         Charset.defaultCharset()
//     );
//     CreateIndexRequestBuilder createIndexRequestBuilder =
//         esClient
//             .admin()
//             .indices()
//             .prepareCreate(INDEX)
//             .setSettings(settings);
//     final String mapping = Resources.toString(
//         getClass().getResource("/elasticsearch/product_mappings.json"),
//         Charset.defaultCharset()
//     );
//     createIndexRequestBuilder = createIndexRequestBuilder.addMapping(TYPE, mapping);
//     return createIndexRequestBuilder.get().isShardsAcked();
//   }
// 
//   /**
//    * Search Product by keywords.
//    *
//    * @param request search request which contains keyword and limit
//    * @return SearchProductsResponse
//    */
//   public SearchProductsResponse searchProducts(SearchProductsRequest request) throws InvalidProtocolBufferException {
//     QueryBuilder queryBuilder = QueryBuilders.boolQuery()
//         .must(QueryBuilders.matchQuery("productName", request.getKeyWord()));
//     SearchResponse response = esClient.prepareSearch(INDEX)
//         .setTypes(TYPE)
//         .setQuery(queryBuilder)
//         .setSize(request.getLimit())
//         .execute()
//         .actionGet();
//     SearchHits hits = response.getHits();
//     SearchProductsResponse.Builder responseBuilder = SearchProductsResponse.newBuilder();
//     for (SearchHit hit : hits) {
//       Product.Builder builder = Product.newBuilder();
//       jsonParser.merge(hit.getSourceAsString(), builder);
//       responseBuilder.addProducts(builder.build());
//     }
//     return responseBuilder.build();
//   }
// 
//   /**
//    * Download product from given category.
//    *
//    * @param request               which contains query category
//    * @param productPublishSubject the subject which downloaded product should publish to
//    */
//   public void downloadProducts(DownloadProductsRequest request, PublishSubject<Product> productPublishSubject) {
// 
//     QueryBuilder queryBuilder = QueryBuilders.termQuery("category", request.getCategory());
//     SearchResponse scrollResponse =
//         esClient
//             .prepareSearch(INDEX)
//             .setScroll(DEFAULT_SCROLL_TIME_VALUE)
//             .setTypes(TYPE)
//             .setQuery(queryBuilder)
//             .setSize(SCROLL_SIZE)
//             .get();
//     do {
//       scrollResponse.getHits().forEach(hit -> {
//         try {
//           Product.Builder builder = Product.newBuilder();
//           jsonParser.merge(hit.sourceAsString(), builder);
//           productPublishSubject.onNext(builder.build());
//         } catch (IOException ioe) {
//           // Don't fail the whole stream
//           log.error("Unable to read product record", ioe);
//           productPublishSubject.onError(ioe);
//           throw new IllegalStateException(ioe);
//         }
//       });
//       // Fetch next batch of cite group records
//       scrollResponse =
//           esClient
//               .prepareSearchScroll(scrollResponse.getScrollId())
//               .setScroll(DEFAULT_SCROLL_TIME_VALUE)
//               .execute()
//               .actionGet();
//     } while (scrollResponse.getHits().getHits().length != 0);
// 
//     productPublishSubject.onComplete();
//   }
// 
//   /**
//    * Calculate product score based on product price for demo only.
//    *
//    * @param product        given product to be calculated
//    * @param downloadStream the download stream for score
//    */
//   public void calculateProductScore(Product product, PublishSubject<CalculateProductScoreResponse> downloadStream) {
//     downloadStream.onNext(
//         CalculateProductScoreResponse
//             .newBuilder()
//             .setProduct(product)
//             .setScore((long) product.getProductPrice())
//             .build()
//     );
//   }
// }


import com.google.inject.Inject;
import com.google.inject.Singleton;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.datanerd.es.dao.ProductDao;
import io.datanerd.generated.common.Product;
import io.datanerd.generated.es.ProductUpdateServiceGrpc;
import io.datanerd.generated.es.UploadProductResponse;
import io.grpc.stub.StreamObserver;
import io.reactivex.subjects.PublishSubject;
package io.datanerd.es.service;




/**
 * Provide Product mutation related service.
 *
 * @author email2liyang@gmail.com
 */
@Singleton
public class ProductUpdateService extends ProductUpdateServiceGrpc.ProductUpdateServiceImplBase {

  private static Logger log = LoggerFactory.getLogger(ProductUpdateService.class); //NOPMD
  @Inject
// Path: grpc-mate-java/elasticsearch-service/src/main/java/io/datanerd/es/dao/ProductDao.java
// @Singleton
// public class ProductDao {
// 
//   private static Logger log = LoggerFactory.getLogger(ProductDao.class); //NOPMD
//   public static final String INDEX = "products";
//   public static final String TYPE = "Product";
//   public static final TimeValue DEFAULT_SCROLL_TIME_VALUE = new TimeValue(60000);
//   static final int SCROLL_SIZE = 50;
// 
//   @Inject
//   private TransportClient esClient;
// 
//   @Inject
//   private JsonFormat.Printer jsonPrinter;
// 
//   @Inject
//   private JsonFormat.Parser jsonParser;
// 
//   /**
//    * Upsert the given product into ES index.
//    *
//    * @param product given product object
//    */
//   public void upsertProduct(Product product) throws InvalidProtocolBufferException {
//     log.debug("save product into ES");
//     final UpdateRequestBuilder updateRequestBuilder =
//         esClient
//             .prepareUpdate(INDEX, TYPE, String.valueOf(product.getProductId()))
//             .setDoc(
//                 jsonPrinter
//                     .includingDefaultValueFields()
//                     .omittingInsignificantWhitespace()
//                     .print(product)
//             )
//             .setDocAsUpsert(true);
//     updateRequestBuilder.get();
//   }
// 
//   /**
//    * Init index setting and mapping.
//    *
//    * @return true if a new index was created, false otherwise
//    */
//   public boolean initIndexIfNotExists() throws IOException {
//     final IndicesExistsResponse existsResponse = esClient.admin().indices().prepareExists(INDEX).get();
//     if (existsResponse.isExists()) {
//       return false;
//     }
//     final String settings = Resources.toString(
//         getClass().getResource("/elasticsearch/product_settings.json"),
//         Charset.defaultCharset()
//     );
//     CreateIndexRequestBuilder createIndexRequestBuilder =
//         esClient
//             .admin()
//             .indices()
//             .prepareCreate(INDEX)
//             .setSettings(settings);
//     final String mapping = Resources.toString(
//         getClass().getResource("/elasticsearch/product_mappings.json"),
//         Charset.defaultCharset()
//     );
//     createIndexRequestBuilder = createIndexRequestBuilder.addMapping(TYPE, mapping);
//     return createIndexRequestBuilder.get().isShardsAcked();
//   }
// 
//   /**
//    * Search Product by keywords.
//    *
//    * @param request search request which contains keyword and limit
//    * @return SearchProductsResponse
//    */
//   public SearchProductsResponse searchProducts(SearchProductsRequest request) throws InvalidProtocolBufferException {
//     QueryBuilder queryBuilder = QueryBuilders.boolQuery()
//         .must(QueryBuilders.matchQuery("productName", request.getKeyWord()));
//     SearchResponse response = esClient.prepareSearch(INDEX)
//         .setTypes(TYPE)
//         .setQuery(queryBuilder)
//         .setSize(request.getLimit())
//         .execute()
//         .actionGet();
//     SearchHits hits = response.getHits();
//     SearchProductsResponse.Builder responseBuilder = SearchProductsResponse.newBuilder();
//     for (SearchHit hit : hits) {
//       Product.Builder builder = Product.newBuilder();
//       jsonParser.merge(hit.getSourceAsString(), builder);
//       responseBuilder.addProducts(builder.build());
//     }
//     return responseBuilder.build();
//   }
// 
//   /**
//    * Download product from given category.
//    *
//    * @param request               which contains query category
//    * @param productPublishSubject the subject which downloaded product should publish to
//    */
//   public void downloadProducts(DownloadProductsRequest request, PublishSubject<Product> productPublishSubject) {
// 
//     QueryBuilder queryBuilder = QueryBuilders.termQuery("category", request.getCategory());
//     SearchResponse scrollResponse =
//         esClient
//             .prepareSearch(INDEX)
//             .setScroll(DEFAULT_SCROLL_TIME_VALUE)
//             .setTypes(TYPE)
//             .setQuery(queryBuilder)
//             .setSize(SCROLL_SIZE)
//             .get();
//     do {
//       scrollResponse.getHits().forEach(hit -> {
//         try {
//           Product.Builder builder = Product.newBuilder();
//           jsonParser.merge(hit.sourceAsString(), builder);
//           productPublishSubject.onNext(builder.build());
//         } catch (IOException ioe) {
//           // Don't fail the whole stream
//           log.error("Unable to read product record", ioe);
//           productPublishSubject.onError(ioe);
//           throw new IllegalStateException(ioe);
//         }
//       });
//       // Fetch next batch of cite group records
//       scrollResponse =
//           esClient
//               .prepareSearchScroll(scrollResponse.getScrollId())
//               .setScroll(DEFAULT_SCROLL_TIME_VALUE)
//               .execute()
//               .actionGet();
//     } while (scrollResponse.getHits().getHits().length != 0);
// 
//     productPublishSubject.onComplete();
//   }
// 
//   /**
//    * Calculate product score based on product price for demo only.
//    *
//    * @param product        given product to be calculated
//    * @param downloadStream the download stream for score
//    */
//   public void calculateProductScore(Product product, PublishSubject<CalculateProductScoreResponse> downloadStream) {
//     downloadStream.onNext(
//         CalculateProductScoreResponse
//             .newBuilder()
//             .setProduct(product)
//             .setScore((long) product.getProductPrice())
//             .build()
//     );
//   }
// }


// Path: grpc-mate-java/elasticsearch-service/src/main/java/io/datanerd/es/service/ProductUpdateService.java
import com.google.inject.Inject;
import com.google.inject.Singleton;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.datanerd.es.dao.ProductDao;
import io.datanerd.generated.common.Product;
import io.datanerd.generated.es.ProductUpdateServiceGrpc;
import io.datanerd.generated.es.UploadProductResponse;
import io.grpc.stub.StreamObserver;
import io.reactivex.subjects.PublishSubject;

package io.datanerd.es.service;




/**
 * Provide Product mutation related service.
 *
 * @author email2liyang@gmail.com
 */
@Singleton
public class ProductUpdateService extends ProductUpdateServiceGrpc.ProductUpdateServiceImplBase {

  private static Logger log = LoggerFactory.getLogger(ProductUpdateService.class); //NOPMD
  @Inject
  private ProductDao productDao;