Skip to content

P
Product Producer
Commit 755fd443 authored by Giridhari Sahoo's avatar Giridhari Sahoo
Browse files
Options

initial commit

parents
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 574 additions and 0 deletions
.gitignore 0 → 100644
View file @ 755fd443
HELP.md
target/
!.mvn/wrapper/maven-wrapper.jar
!**/src/main/**/target/
!**/src/test/**/target/
### STS ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
.sts4-cache
### IntelliJ IDEA ###
.idea
*.iws
*.iml
*.ipr
### NetBeans ###
/nbproject/private/
/nbbuild/
/dist/
/nbdist/
/.nb-gradle/
build/
!**/src/main/**/build/
!**/src/test/**/build/
### VS Code ###
.vscode/
.mvn/wrapper/maven-wrapper.properties 0 → 100644
View file @ 755fd443
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
wrapperVersion=3.3.2
distributionType=only-script
distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.9/apache-maven-3.9.9-bin.zip
mvnw 0 → 100644
View file @ 755fd443
This diff is collapsed.
mvnw.cmd 0 → 100644
View file @ 755fd443
<# : batch portion
@REM ----------------------------------------------------------------------------
@REM Licensed to the Apache Software Foundation (ASF) under one
@REM or more contributor license agreements. See the NOTICE file
@REM distributed with this work for additional information
@REM regarding copyright ownership. The ASF licenses this file
@REM to you under the Apache License, Version 2.0 (the
@REM "License"); you may not use this file except in compliance
@REM with the License. You may obtain a copy of the License at
@REM
@REM http://www.apache.org/licenses/LICENSE-2.0
@REM
@REM Unless required by applicable law or agreed to in writing,
@REM software distributed under the License is distributed on an
@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@REM KIND, either express or implied. See the License for the
@REM specific language governing permissions and limitations
@REM under the License.
@REM ----------------------------------------------------------------------------
@REM ----------------------------------------------------------------------------
@REM Apache Maven Wrapper startup batch script, version 3.3.2
@REM
@REM Optional ENV vars
@REM MVNW_REPOURL - repo url base for downloading maven distribution
@REM MVNW_USERNAME/MVNW_PASSWORD - user and password for downloading maven
@REM MVNW_VERBOSE - true: enable verbose log; others: silence the output
@REM ----------------------------------------------------------------------------
@IF "%__MVNW_ARG0_NAME__%"=="" (SET __MVNW_ARG0_NAME__=%~nx0)
@SET __MVNW_CMD__=
@SET __MVNW_ERROR__=
@SET __MVNW_PSMODULEP_SAVE=%PSModulePath%
@SET PSModulePath=
@FOR /F "usebackq tokens=1* delims==" %%A IN (`powershell -noprofile "& {$scriptDir='%~dp0'; $script='%__MVNW_ARG0_NAME__%'; icm -ScriptBlock ([Scriptblock]::Create((Get-Content -Raw '%~f0'))) -NoNewScope}"`) DO @(
IF "%%A"=="MVN_CMD" (set __MVNW_CMD__=%%B) ELSE IF "%%B"=="" (echo %%A) ELSE (echo %%A=%%B)
)
@SET PSModulePath=%__MVNW_PSMODULEP_SAVE%
@SET __MVNW_PSMODULEP_SAVE=
@SET __MVNW_ARG0_NAME__=
@SET MVNW_USERNAME=
@SET MVNW_PASSWORD=
@IF NOT "%__MVNW_CMD__%"=="" (%__MVNW_CMD__% %*)
@echo Cannot start maven from wrapper >&2 && exit /b 1
@GOTO :EOF
: end batch / begin powershell #>
$ErrorActionPreference = "Stop"
if ($env:MVNW_VERBOSE -eq "true") {
$VerbosePreference = "Continue"
}
# calculate distributionUrl, requires .mvn/wrapper/maven-wrapper.properties
$distributionUrl = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionUrl
if (!$distributionUrl) {
Write-Error "cannot read distributionUrl property in $scriptDir/.mvn/wrapper/maven-wrapper.properties"
}
switch -wildcard -casesensitive ( $($distributionUrl -replace '^.*/','') ) {
"maven-mvnd-*" {
$USE_MVND = $true
$distributionUrl = $distributionUrl -replace '-bin\.[^.]*$',"-windows-amd64.zip"
$MVN_CMD = "mvnd.cmd"
break
}
default {
$USE_MVND = $false
$MVN_CMD = $script -replace '^mvnw','mvn'
break
}
}
# apply MVNW_REPOURL and calculate MAVEN_HOME
# maven home pattern: ~/.m2/wrapper/dists/{apache-maven-<version>,maven-mvnd-<version>-<platform>}/<hash>
if ($env:MVNW_REPOURL) {
$MVNW_REPO_PATTERN = if ($USE_MVND) { "/org/apache/maven/" } else { "/maven/mvnd/" }
$distributionUrl = "$env:MVNW_REPOURL$MVNW_REPO_PATTERN$($distributionUrl -replace '^.*'+$MVNW_REPO_PATTERN,'')"
}
$distributionUrlName = $distributionUrl -replace '^.*/',''
$distributionUrlNameMain = $distributionUrlName -replace '\.[^.]*$','' -replace '-bin$',''
$MAVEN_HOME_PARENT = "$HOME/.m2/wrapper/dists/$distributionUrlNameMain"
if ($env:MAVEN_USER_HOME) {
$MAVEN_HOME_PARENT = "$env:MAVEN_USER_HOME/wrapper/dists/$distributionUrlNameMain"
}
$MAVEN_HOME_NAME = ([System.Security.Cryptography.MD5]::Create().ComputeHash([byte[]][char[]]$distributionUrl) | ForEach-Object {$_.ToString("x2")}) -join ''
$MAVEN_HOME = "$MAVEN_HOME_PARENT/$MAVEN_HOME_NAME"
if (Test-Path -Path "$MAVEN_HOME" -PathType Container) {
Write-Verbose "found existing MAVEN_HOME at $MAVEN_HOME"
Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
exit $?
}
if (! $distributionUrlNameMain -or ($distributionUrlName -eq $distributionUrlNameMain)) {
Write-Error "distributionUrl is not valid, must end with *-bin.zip, but found $distributionUrl"
}
# prepare tmp dir
$TMP_DOWNLOAD_DIR_HOLDER = New-TemporaryFile
$TMP_DOWNLOAD_DIR = New-Item -Itemtype Directory -Path "$TMP_DOWNLOAD_DIR_HOLDER.dir"
$TMP_DOWNLOAD_DIR_HOLDER.Delete() | Out-Null
trap {
if ($TMP_DOWNLOAD_DIR.Exists) {
try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
}
}
New-Item -Itemtype Directory -Path "$MAVEN_HOME_PARENT" -Force | Out-Null
# Download and Install Apache Maven
Write-Verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
Write-Verbose "Downloading from: $distributionUrl"
Write-Verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
$webclient = New-Object System.Net.WebClient
if ($env:MVNW_USERNAME -and $env:MVNW_PASSWORD) {
$webclient.Credentials = New-Object System.Net.NetworkCredential($env:MVNW_USERNAME, $env:MVNW_PASSWORD)
}
[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
$webclient.DownloadFile($distributionUrl, "$TMP_DOWNLOAD_DIR/$distributionUrlName") | Out-Null
# If specified, validate the SHA-256 sum of the Maven distribution zip file
$distributionSha256Sum = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionSha256Sum
if ($distributionSha256Sum) {
if ($USE_MVND) {
Write-Error "Checksum validation is not supported for maven-mvnd. `nPlease disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties."
}
Import-Module $PSHOME\Modules\Microsoft.PowerShell.Utility -Function Get-FileHash
if ((Get-FileHash "$TMP_DOWNLOAD_DIR/$distributionUrlName" -Algorithm SHA256).Hash.ToLower() -ne $distributionSha256Sum) {
Write-Error "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised. If you updated your Maven version, you need to update the specified distributionSha256Sum property."
}
}
# unzip and move
Expand-Archive "$TMP_DOWNLOAD_DIR/$distributionUrlName" -DestinationPath "$TMP_DOWNLOAD_DIR" | Out-Null
Rename-Item -Path "$TMP_DOWNLOAD_DIR/$distributionUrlNameMain" -NewName $MAVEN_HOME_NAME | Out-Null
try {
Move-Item -Path "$TMP_DOWNLOAD_DIR/$MAVEN_HOME_NAME" -Destination $MAVEN_HOME_PARENT | Out-Null
} catch {
if (! (Test-Path -Path "$MAVEN_HOME" -PathType Container)) {
Write-Error "fail to move MAVEN_HOME"
}
} finally {
try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
}
Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
pom.xml 0 → 100644
View file @ 755fd443
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.3.4</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.nisum</groupId>
<artifactId>productreader</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>productreader</name>
<description>This project is about read product information from csv file and publish it into kafka</description>
<url/>
<licenses>
<license/>
</licenses>
<developers>
<developer/>
</developers>
<scm>
<connection/>
<developerConnection/>
<tag/>
<url/>
</scm>
<properties>
<java.version>17</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/com.opencsv/opencsv -->
<dependency>
<groupId>com.opencsv</groupId>
<artifactId>opencsv</artifactId>
<version>5.7.1</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
src/main/java/com/nisum/productreader/ProductreaderApplication.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ProductreaderApplication {
public static void main(String[] args) {
SpringApplication.run(ProductreaderApplication.class, args);
}
}
src/main/java/com/nisum/productreader/config/KafkaConfig.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader.config;
import com.nisum.productreader.model.Product;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import java.util.HashMap;
import java.util.Map;
public class KafkaConfig {
@Bean
public ProducerFactory<String, Product> producerFactory() {
Map<String, Object> config = new HashMap<>();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
return new DefaultKafkaProducerFactory<>(config);
}
@Bean
public KafkaTemplate<String, Product> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
src/main/java/com/nisum/productreader/controller/ProductController.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader.controller;
import com.nisum.productreader.model.Product;
import com.nisum.productreader.service.ProductKafkaProducer;
import com.nisum.productreader.service.ProductService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.util.Loggers;
@RestController
@RequestMapping("/products")
public class ProductController {
private Logger logger = LoggerFactory.getLogger(ProductController.class);
@Autowired
private ProductService productService;
@Autowired
private ProductKafkaProducer kafkaProducer;
@GetMapping("/getAllProducts")
public Flux<Product> getAllProducts(){
logger.info("Received request to get all products from CSV file.");
Flux<Product> products = productService.readProductFromCsv();
// Publish the products to Kafka topic
kafkaProducer.publishProducts(products, "product-topic"); // Replace with your topic name
return products;
}
}
src/main/java/com/nisum/productreader/model/Product.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader.model;
import java.util.List;
public class Product {
private Integer productId;
private String productName;
private String productCategory;
private List<Double> priceList;
private String color;
private String ram;
private String model;
public Integer getProductId() {
return productId;
}
public void setProductId(Integer productId) {
this.productId = productId;
}
public String getProductName() {
return productName;
}
public void setProductName(String productName) {
this.productName = productName;
}
public String getProductCategory() {
return productCategory;
}
public void setProductCategory(String productCategory) {
this.productCategory = productCategory;
}
public List<Double> getPriceList() {
return priceList;
}
public void setPriceList(List<Double> priceList) {
this.priceList = priceList;
}
public String getColor() {
return color;
}
public void setColor(String color) {
this.color = color;
}
public String getRam() {
return ram;
}
public void setRam(String ram) {
this.ram = ram;
}
public String getModel() {
return model;
}
public void setModel(String model) {
this.model = model;
}
@Override
public String toString() {
return "Product{" +
"productId=" + productId +
", productName='" + productName + '\'' +
", productCategory='" + productCategory + '\'' +
", priceList=" + priceList +
", color='" + color + '\'' +
", ram='" + ram + '\'' +
", model='" + model + '\'' +
'}';
}
}
src/main/java/com/nisum/productreader/service/ProductKafkaProducer.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader.service;
import com.nisum.productreader.model.Product;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
@Service
public class ProductKafkaProducer {
private static final Logger logger = LoggerFactory.getLogger(ProductKafkaProducer.class);
private final KafkaTemplate<String, Product> kafkaTemplate;
@Autowired
public ProductKafkaProducer(KafkaTemplate<String, Product> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void publishProducts(Flux<Product> products, String topic) {
products.doOnNext(product -> {
kafkaTemplate.send(topic, product.getProductId().toString(), product); // Send product
logger.info("Published product to Kafka: {}", product);
}).subscribe(); // Subscribe to initiate the Flux processing
}
}
src/main/java/com/nisum/productreader/service/ProductService.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader.service;
import com.nisum.productreader.model.Product;
import com.opencsv.CSVReader;
import com.opencsv.exceptions.CsvException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Arrays;
import java.util.List;
@Service
public class ProductService {
private static final Logger logger = LoggerFactory.getLogger(ProductService.class);
public Flux<Product> readProductFromCsv() {
logger.info("Starting to read the CSV file from classpath");
return Flux.create(sink -> {
try {
// Load CSV file from classpath
ClassPathResource resource = new ClassPathResource("product.csv");
CSVReader csvReader = new CSVReader(new InputStreamReader(resource.getInputStream()));
List<String[]> rows = csvReader.readAll();
logger.info("Successfully read {} rows from the CSV file", rows.size() - 1);
rows.stream().skip(1).forEach(row -> {
Product product = new Product();
product.setProductId(Integer.parseInt(row[0]));
product.setProductName(row[1]);
product.setProductCategory(row[2]);
// Convert list prices into List<Double>
List<Double> prices = Arrays.asList(row[3].split(","))
.stream().map(String::trim)
.map(Double::parseDouble).toList();
product.setPriceList(prices);
product.setColor(row[4]);
product.setRam(row[5]);
product.setModel(row[6]);
logger.info("Parsed product: {}", product); // Log product details at debug level
sink.next(product); // Emit product
});
sink.complete(); // Complete the flux
logger.info("Completed reading and processing the CSV file.");
} catch (Exception e) {
logger.error("Error reading CSV file: {}", e.getMessage(), e);
sink.error(e); // Error signal if exception occurs
}
});
}
}
src/main/java/com/nisum/productreader/service/TestProductRead.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader.service;
public class TestProductRead {
// public static void main(String[] args) {
// ProductService productService = new ProductService();
// productService.readProductFromCsv().subscribe();
// }
}
src/main/resources/application.properties 0 → 100644
View file @ 755fd443
spring.application.name=productreader
server.port=8081
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
\ No newline at end of file
src/main/resources/product.csv 0 → 100644
View file @ 755fd443
Product ID,Product Name,Product Category,List Price,Color,Ram,Model
1001,iPhone 14,Mobile Phone,"7000.00, 5500.00, 4000.00",black,8gb,"A15 Bionic chip"
1002,Dell XPS 13,Laptop,"12000.00, 10800.00, 8000.00",gray,16gb,Touchscreen
1003,Samsung Galaxy s21,Mobile Phone,"8000.00, 6800.00, 6000.00",white,8gb,Exynos 2100
1004,Hp Specter x360,Laptop,"14000.00, 12500.00, 11000.00",blue,16gb,convertible
\ No newline at end of file
src/test/java/com/nisum/productreader/ProductreaderApplicationTests.java 0 → 100644
View file @ 755fd443
package com.nisum.productreader;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class ProductreaderApplicationTests {
@Test
void contextLoads() {
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment