Skip to content

F
FMS_Project
You need to sign in or sign up before continuing.
Unverified Commit be1cb6b5 authored by Basharo77's avatar Basharo77 Committed by GitHub
Browse files
Options

Add files via upload

parent b705a2c6
Expand all Show whitespace changes
Inline Side-by-side
Showing with 2319 additions and 0 deletions
First_Pipeline/consumer/mvnw 0 → 100644
View file @ be1cb6b5
This diff is collapsed.
First_Pipeline/consumer/mvnw.cmd 0 → 100644
View file @ be1cb6b5
@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 https://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 Maven Start Up Batch script
@REM
@REM Required ENV vars:
@REM JAVA_HOME - location of a JDK home dir
@REM
@REM Optional ENV vars
@REM M2_HOME - location of maven2's installed home dir
@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a keystroke before ending
@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
@REM e.g. to debug Maven itself, use
@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
@REM ----------------------------------------------------------------------------
@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
@echo off
@REM set title of command window
title %0
@REM enable echoing by setting MAVEN_BATCH_ECHO to 'on'
@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO%
@REM set %HOME% to equivalent of $HOME
if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
@REM Execute a user defined script before this one
if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
@REM check for pre script, once with legacy .bat ending and once with .cmd ending
if exist "%USERPROFILE%\mavenrc_pre.bat" call "%USERPROFILE%\mavenrc_pre.bat" %*
if exist "%USERPROFILE%\mavenrc_pre.cmd" call "%USERPROFILE%\mavenrc_pre.cmd" %*
:skipRcPre
@setlocal
set ERROR_CODE=0
@REM To isolate internal variables from possible post scripts, we use another setlocal
@setlocal
@REM ==== START VALIDATION ====
if not "%JAVA_HOME%" == "" goto OkJHome
echo.
echo Error: JAVA_HOME not found in your environment. >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
:OkJHome
if exist "%JAVA_HOME%\bin\java.exe" goto init
echo.
echo Error: JAVA_HOME is set to an invalid directory. >&2
echo JAVA_HOME = "%JAVA_HOME%" >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
@REM ==== END VALIDATION ====
:init
@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
@REM Fallback to current working directory if not found.
set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
set EXEC_DIR=%CD%
set WDIR=%EXEC_DIR%
:findBaseDir
IF EXIST "%WDIR%"\.mvn goto baseDirFound
cd ..
IF "%WDIR%"=="%CD%" goto baseDirNotFound
set WDIR=%CD%
goto findBaseDir
:baseDirFound
set MAVEN_PROJECTBASEDIR=%WDIR%
cd "%EXEC_DIR%"
goto endDetectBaseDir
:baseDirNotFound
set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
cd "%EXEC_DIR%"
:endDetectBaseDir
IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
@setlocal EnableExtensions EnableDelayedExpansion
for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%
:endReadAdditionalConfig
SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
set DOWNLOAD_URL="https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
FOR /F "usebackq tokens=1,2 delims==" %%A IN ("%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties") DO (
IF "%%A"=="wrapperUrl" SET DOWNLOAD_URL=%%B
)
@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
@REM This allows using the maven wrapper in projects that prohibit checking in binary data.
if exist %WRAPPER_JAR% (
if "%MVNW_VERBOSE%" == "true" (
echo Found %WRAPPER_JAR%
)
) else (
if not "%MVNW_REPOURL%" == "" (
SET DOWNLOAD_URL="%MVNW_REPOURL%/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
)
if "%MVNW_VERBOSE%" == "true" (
echo Couldn't find %WRAPPER_JAR%, downloading it ...
echo Downloading from: %DOWNLOAD_URL%
)
powershell -Command "&{"^
"$webclient = new-object System.Net.WebClient;"^
"if (-not ([string]::IsNullOrEmpty('%MVNW_USERNAME%') -and [string]::IsNullOrEmpty('%MVNW_PASSWORD%'))) {"^
"$webclient.Credentials = new-object System.Net.NetworkCredential('%MVNW_USERNAME%', '%MVNW_PASSWORD%');"^
"}"^
"[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12; $webclient.DownloadFile('%DOWNLOAD_URL%', '%WRAPPER_JAR%')"^
"}"
if "%MVNW_VERBOSE%" == "true" (
echo Finished downloading %WRAPPER_JAR%
)
)
@REM End of extension
@REM Provide a "standardized" way to retrieve the CLI args that will
@REM work with both Windows and non-Windows executions.
set MAVEN_CMD_LINE_ARGS=%*
%MAVEN_JAVA_EXE% ^
%JVM_CONFIG_MAVEN_PROPS% ^
%MAVEN_OPTS% ^
%MAVEN_DEBUG_OPTS% ^
-classpath %WRAPPER_JAR% ^
"-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" ^
%WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
if ERRORLEVEL 1 goto error
goto end
:error
set ERROR_CODE=1
:end
@endlocal & set ERROR_CODE=%ERROR_CODE%
if not "%MAVEN_SKIP_RC%"=="" goto skipRcPost
@REM check for post script, once with legacy .bat ending and once with .cmd ending
if exist "%USERPROFILE%\mavenrc_post.bat" call "%USERPROFILE%\mavenrc_post.bat"
if exist "%USERPROFILE%\mavenrc_post.cmd" call "%USERPROFILE%\mavenrc_post.cmd"
:skipRcPost
@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
if "%MAVEN_BATCH_PAUSE%"=="on" pause
if "%MAVEN_TERMINATE_CMD%"=="on" exit %ERROR_CODE%
cmd /C exit /B %ERROR_CODE%
First_Pipeline/consumer/pom.xml 0 → 100644
View file @ be1cb6b5
<?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.0.4</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>consumer</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>consumer</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>17</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</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>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.rethinkdb</groupId>
<artifactId>rethinkdb-driver</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.8.9</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-messaging</artifactId>
<version>5.3.9</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.8.6</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>javax.persistence</groupId>
<artifactId>javax.persistence-api</artifactId>
<version>2.2</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
First_Pipeline/consumer/src/main/java/com/example/consumer/ConsumerApplication.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer;
import com.example.consumer.configuration.SocketTextHandler;
import com.example.consumer.factory.RethinkDBConnectionFactory;
import com.example.consumer.repository.RethinkChange;
import com.rethinkdb.RethinkDB;
import com.rethinkdb.net.Connection;
import com.rethinkdb.net.Cursor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@SpringBootApplication
public class ConsumerApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(ConsumerApplication.class, args);
}
private final RethinkDB r = RethinkDB.r;
@Autowired
public RethinkDBConnectionFactory connectionFactory;
@Autowired
private SocketTextHandler socket_server;
@Override
public void run(String... args) throws Exception {
Connection connection = connectionFactory.getConnection();
Cursor<RethinkChange> changeCursor = r.db("my_database").table("my_table").changes().optArg("include_initial",true).
run(connection, RethinkChange.class);
List<Map<String, Object>> result = new ArrayList<>();
for (RethinkChange change : changeCursor){
System.out.println("Something Changed");
result.add(change.getNew_val());
socket_server.broadcast(result);
}
}
// @KafkaListener(topics = "Test")
// public void handleNotification(String s) {
//
// System.out.println("Received Message from Producer: "+s+"\n");
// }
}
First_Pipeline/consumer/src/main/java/com/example/consumer/configuration/SocketTextHandler.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.configuration;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
@Component
public class SocketTextHandler extends TextWebSocketHandler {
private List<WebSocketSession> sessions = new CopyOnWriteArrayList<>();
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
System.out.println("New connection from " + session.getRemoteAddress());
sessions.add(session);
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
System.out.println("Received message from " + session.getRemoteAddress() + ": " + message.getPayload());
// Handle incoming message
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
System.out.println("Connection closed to " + session.getRemoteAddress() + " with status " + status);
sessions.remove(session);
}
public void broadcast(List<Map<String, Object>> message) throws IOException {
System.out.println("Hey Broadcasting");
System.out.println(message);
ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter();
String json = ow.writeValueAsString(message);
System.out.println(json);
TextMessage s = new TextMessage(json);
System.out.println(s);
for (WebSocketSession session : sessions) {
if (session.isOpen()) {
session.sendMessage(s);
}
}
}
}
First_Pipeline/consumer/src/main/java/com/example/consumer/configuration/WebSocketConfig.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.configuration;
import com.example.consumer.services.RethinkDBService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.*;
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
@Autowired
private SocketTextHandler socket_server;
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
System.out.println("HEYYYYYYYYYYYYYYYYYY");
registry.addHandler(socket_server, "/my-websocket").setAllowedOrigins("*");
}
}
First_Pipeline/consumer/src/main/java/com/example/consumer/controller/RethinkController.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.controller;
import com.example.consumer.services.RethinkDBService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.*;
import java.util.List;
import java.util.Map;
@RestController
@CrossOrigin(origins = {"http://localhost:5173"})
@RequestMapping("/api/rethink")
public class RethinkController {
@Autowired
private RethinkDBService rethinkDBService;
@PostMapping("/data")
public ResponseEntity<String> saveData(@RequestBody Map<String, Object> data) {
rethinkDBService.saveData("my_database", "my_table", data);
return ResponseEntity.ok("Data saved successfully in RethinkDB");
}
@GetMapping("/data")
public ResponseEntity<List<Map<String, Object>>> getData() {
List<Map<String, Object>> result = rethinkDBService.getData("my_database", "my_table");
System.out.println(result);
if (result != null) {
return ResponseEntity.ok(result);
} else {
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
}
}
@KafkaListener(topics = "Test")
public void handleKafkaMessage(String message) {
rethinkDBService.saveKafkaMessageToRethink(message);
}
}
First_Pipeline/consumer/src/main/java/com/example/consumer/controller/TrapController.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.controller;
import com.example.consumer.entity.Trap;
import com.example.consumer.services.TrapService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.*;
@RestController
@CrossOrigin(origins = {"http://localhost:5173"})
@RequestMapping("/api/elastic")
public class TrapController {
@Autowired
private TrapService trapService;
@GetMapping("/findAllTraps")
Iterable<Trap> findAll(){
return trapService.getTraps();
}
@PostMapping("/insertTraps")
public Trap insertTrap(@RequestBody Trap trap){
return trapService.insertTrap(trap);
}
@KafkaListener(topics = "Test")
public void handleKafkaMessage(String message) {
trapService.saveKafkaMessageToElastic(message);
}
}
First_Pipeline/consumer/src/main/java/com/example/consumer/entity/Trap.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.entity;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
@NoArgsConstructor
@Data
@Document(indexName = "traps")
public class Trap {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private String id;
private String trap;
public Trap(String trap) {
this.trap = trap;
}
}
First_Pipeline/consumer/src/main/java/com/example/consumer/factory/RethinkDBConnectionFactory.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.factory;
import com.rethinkdb.RethinkDB;
import com.rethinkdb.net.Connection;
import com.rethinkdb.net.Cursor;
import jakarta.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;
@Component
public class RethinkDBConnectionFactory {
private final Logger log = LoggerFactory.getLogger(RethinkDBConnectionFactory.class);
private final RethinkDB r = RethinkDB.r;
private Connection connection;
@Value("${rethinkdb.host}")
private String host;
@Value("${rethinkdb.port}")
private int port;
@PostConstruct
public Connection init() {
try {
connection = r.connection().hostname(host).port(port).connect();
log.info("RethinkDB connected successfully");
List<String> dbList = r.dbList().run(connection);
if (!dbList.contains("my_database")) {
System.out.println("Creating DATABASE Heeeeeeeeeeeeeeeeeeeeeeeeeere");
r.dbCreate("my_database").run(connection);
}
List<String> tables = r.db("my_database").tableList().run(connection);
if (!tables.contains("my_table")) {
System.out.println("Creating Table Heeeeeeeeeeeeeeeeeeeeeeeeeere");
r.db("my_database").tableCreate("my_table").run(connection);
//r.db("my_database").table("my_table").indexCreate("trap").run(connection);
}
} catch (Exception e) {
log.error("Error connecting to RethinkDB", e);
}
return null;
}
public Connection getConnection() {
return connection;
}
}
First_Pipeline/consumer/src/main/java/com/example/consumer/repository/RethinkChange.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.repository;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Map;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class RethinkChange {
private Map<String,Object> new_val;
private Map<String,Object> old_val;
}
First_Pipeline/consumer/src/main/java/com/example/consumer/repository/TrapRepository.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.repository;
import com.example.consumer.entity.Trap;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;
@Repository
public interface TrapRepository extends ElasticsearchRepository<Trap,Integer> {
}
First_Pipeline/consumer/src/main/java/com/example/consumer/services/DbInitializer.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.services;
import com.example.consumer.factory.RethinkDBConnectionFactory;
import com.rethinkdb.RethinkDB;
import com.rethinkdb.net.Connection;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.List;
public class DbInitializer implements InitializingBean {
@Autowired
private RethinkDBConnectionFactory connectionFactory;
@Autowired
private RethinkDBService rethinkDBService;
private static final RethinkDB r = RethinkDB.r;
@Override
public void afterPropertiesSet() throws Exception {
createDb();
}
private void createDb() {
System.out.println("We are initializing heeeeeeeeeeeeeeeeeeeeeer");
Connection connection = connectionFactory.getConnection();
List<String> dbList = r.dbList().run(connection);
if (!dbList.contains("new_database")) {
System.out.println("Creating DATABASE Heeeeeeeeeeeeeeeeeeeeeeeeeere");
r.dbCreate("new_database").run(connection);
}
List<String> tables = r.db("new_database").tableList().run(connection);
if (!tables.contains("new_table")) {
System.out.println("Creating Table Heeeeeeeeeeeeeeeeeeeeeeeeeere");
r.db("new_database").tableCreate("new_table").run(connection);
r.db("new_database").table("new_table").indexCreate("trap").run(connection);
}
}
}
\ No newline at end of file
First_Pipeline/consumer/src/main/java/com/example/consumer/services/RethinkDBService.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.services;
import com.example.consumer.configuration.SocketTextHandler;
import com.example.consumer.factory.RethinkDBConnectionFactory;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rethinkdb.RethinkDB;
import com.rethinkdb.gen.ast.Json;
import com.rethinkdb.gen.ast.Table;
import com.rethinkdb.net.Cursor;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import com.rethinkdb.net.Connection;
import com.rethinkdb.model.MapObject;
import org.springframework.web.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.core.publisher.Sinks.Many;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
@Service
public class RethinkDBService {
private final Logger log = LoggerFactory.getLogger(RethinkDBService.class);
private final RethinkDB r = RethinkDB.r;
@Autowired
public RethinkDBConnectionFactory connectionFactory;
public void saveKafkaMessageToRethink(String message) {
try {
ObjectMapper objectMapper = new ObjectMapper();
String jsonString = "{\"trap\":\" "+message+"\"}";
JsonNode jsonNode = objectMapper.readTree(jsonString);
// Create a RethinkDB document object using the parsed JSON object
Map<String, Object> document = objectMapper.convertValue(jsonNode, Map.class);
r.db("my_database").table("my_table").insert(document).run(connectionFactory.getConnection());
} catch (IOException e) {
System.out.println("error " + e);
}
}
public void saveData(String database, String table, Map<String, Object> data) {
try {
r.db(database).table(table).insert(data).run(connectionFactory.getConnection());
log.info("Data saved successfully in RethinkDB");
} catch (Exception e) {
log.error("Error saving data in RethinkDB", e);
}
}
public List<Map<String, Object>> getData(String database, String table) {
try {
Cursor<Map<String, Object>> cursor = r.db(database).table(table).run(connectionFactory.getConnection());
List<Map<String, Object>> result = new ArrayList<>();
while (cursor.hasNext()) {
result.add(cursor.next());
}
return result;
} catch (Exception e) {
log.error("Error getting data from RethinkDB", e);
return null;
}
}
//Connection conn = connectionFactory.getConnection();
//Cursor<Map<String, Object> > changeCursor = r.db(database).table(table).changes().run(conn);
/*for (Object change : changeCursor) {
result.add((Map<String, Object>) change);
}
try {
socket_server.broadcast(result);
} catch (IOException e) {
throw new RuntimeException(e);
}*/
}
First_Pipeline/consumer/src/main/java/com/example/consumer/services/TrapService.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer.services;
import com.example.consumer.repository.TrapRepository;
import com.example.consumer.entity.Trap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class TrapService {
@Autowired
private TrapRepository trapRepository;
private final Logger log = LoggerFactory.getLogger(TrapService.class);
public Iterable<Trap> getTraps() {
try{
return trapRepository.findAll();
} catch (Exception e) {
log.error("Error getting data from Elastic", e);
return null;
}
}
public Trap insertTrap(Trap trap) {
try {
trapRepository.save(trap);
log.info("Data saved successfully in Elastic");
return trap;
}
catch (Exception e) {
log.error("Error saving data in Elastic", e);
}
return null;
}
public Trap updateTrap(Trap trap, int id) {
Trap trap1 = trapRepository.findById(id).get();
trap1.setTrap(trap.getTrap());
return trap1;
}
public void deleteTrap(int id ) {
trapRepository.deleteById(id);
}
public void saveKafkaMessageToElastic(String message) {
try {
Trap trap = new Trap(message);
trapRepository.save(trap);
log.info("Data saved successfully in Elastic");
} catch (Exception e) {
System.out.println("error " + e);
}
}
}
First_Pipeline/consumer/src/main/resources/application.properties 0 → 100644
View file @ be1cb6b5
server.port = 4164
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.template.default-topic=notificationTopic
spring.kafka.consumer.group-id= notificationId
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
rethinkdb.host = localhost
rethinkdb.port = 28015
spring.data.elasticsearch.cluster-names=trapsService
spring.data.elasticsearch.cluster-node=localhost:9200
First_Pipeline/consumer/src/test/java/com/example/consumer/ConsumerApplicationTests.java 0 → 100644
View file @ be1cb6b5
package com.example.consumer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class ConsumerApplicationTests {
@Test
void contextLoads() {
}
}
First_Pipeline/data/rethinkdb_data/log_file 0 → 100644
View file @ be1cb6b5
This diff is collapsed.
First_Pipeline/docker-compose-elk.yaml 0 → 100644
View file @ be1cb6b5
version: "3"
services:
elasticsearch:
image: elasticsearch:8.7.1
container_name: elasticsearch
environment:
- bootstrap.memory_lock=true
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
- "discovery.type=single-node"
- xpack.security.enabled=false
ports:
- "9200:9200"
volumes:
- elasticsearch_data:/usr/share/elasticsearch/data
networks:
- elastic
kibana:
image: kibana:8.7.1
container_name: kibana
ports:
- "5601:5601"
environment:
ELASTICSEARCH_URL: http://elasticsearch:9200
ELASTICSEARCH_HOSTS: '["http://elasticsearch:9200"]'
depends_on:
- elasticsearch
networks:
- elastic
#
# logstash:
# image: docker.elastic.co/logstash/logstash:8.3.3
# container_name: logstash
# volumes:
# - ./docker/logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml:ro
# - ./docker/logstash/pipeline:/usr/share/logstash/pipeline:ro
# ports:
# - "5044:5044"
# - "5000:5000/tcp"
# - "5000:5000/udp"
# - "9600:9600"
# environment:
# LS_JAVA_OPTS: "-Xmx256m -Xms256m"
# networks:
# - elastic
# depends_on:
# - elasticsearch
#
# zipkin:
# image: openzipkin/zipkin
# container_name: zipkin
# ports:
# - "9411:9411"
# networks:
# - elastic
networks:
elastic:
driver: bridge
volumes:
elasticsearch_data:
driver: local
\ No newline at end of file
First_Pipeline/docker-compose-rethink.yml 0 → 100644
View file @ be1cb6b5
version: '3'
services:
rethinkdb:
image: rethinkdb
ports:
- "8080:8080"
- "28015:28015"
- "29015:29015"
volumes:
- "./data:/data"
First_Pipeline/docker-compose.yml 0 → 100644
View file @ be1cb6b5
version: '3'
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.0.1
container_name: zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
broker:
image: confluentinc/cp-kafka:7.0.1
container_name: broker
ports:
- "9092:9092"
depends_on:
- zookeeper
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://broker:29092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
First_Pipeline/producer/mvnw 0 → 100644
View file @ be1cb6b5
This diff is collapsed.
First_Pipeline/producer/mvnw.cmd 0 → 100644
View file @ be1cb6b5
@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 https://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 Maven Start Up Batch script
@REM
@REM Required ENV vars:
@REM JAVA_HOME - location of a JDK home dir
@REM
@REM Optional ENV vars
@REM M2_HOME - location of maven2's installed home dir
@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a keystroke before ending
@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
@REM e.g. to debug Maven itself, use
@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
@REM ----------------------------------------------------------------------------
@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
@echo off
@REM set title of command window
title %0
@REM enable echoing by setting MAVEN_BATCH_ECHO to 'on'
@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO%
@REM set %HOME% to equivalent of $HOME
if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
@REM Execute a user defined script before this one
if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
@REM check for pre script, once with legacy .bat ending and once with .cmd ending
if exist "%USERPROFILE%\mavenrc_pre.bat" call "%USERPROFILE%\mavenrc_pre.bat" %*
if exist "%USERPROFILE%\mavenrc_pre.cmd" call "%USERPROFILE%\mavenrc_pre.cmd" %*
:skipRcPre
@setlocal
set ERROR_CODE=0
@REM To isolate internal variables from possible post scripts, we use another setlocal
@setlocal
@REM ==== START VALIDATION ====
if not "%JAVA_HOME%" == "" goto OkJHome
echo.
echo Error: JAVA_HOME not found in your environment. >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
:OkJHome
if exist "%JAVA_HOME%\bin\java.exe" goto init
echo.
echo Error: JAVA_HOME is set to an invalid directory. >&2
echo JAVA_HOME = "%JAVA_HOME%" >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
@REM ==== END VALIDATION ====
:init
@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
@REM Fallback to current working directory if not found.
set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
set EXEC_DIR=%CD%
set WDIR=%EXEC_DIR%
:findBaseDir
IF EXIST "%WDIR%"\.mvn goto baseDirFound
cd ..
IF "%WDIR%"=="%CD%" goto baseDirNotFound
set WDIR=%CD%
goto findBaseDir
:baseDirFound
set MAVEN_PROJECTBASEDIR=%WDIR%
cd "%EXEC_DIR%"
goto endDetectBaseDir
:baseDirNotFound
set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
cd "%EXEC_DIR%"
:endDetectBaseDir
IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
@setlocal EnableExtensions EnableDelayedExpansion
for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%
:endReadAdditionalConfig
SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
set DOWNLOAD_URL="https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
FOR /F "usebackq tokens=1,2 delims==" %%A IN ("%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties") DO (
IF "%%A"=="wrapperUrl" SET DOWNLOAD_URL=%%B
)
@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
@REM This allows using the maven wrapper in projects that prohibit checking in binary data.
if exist %WRAPPER_JAR% (
if "%MVNW_VERBOSE%" == "true" (
echo Found %WRAPPER_JAR%
)
) else (
if not "%MVNW_REPOURL%" == "" (
SET DOWNLOAD_URL="%MVNW_REPOURL%/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
)
if "%MVNW_VERBOSE%" == "true" (
echo Couldn't find %WRAPPER_JAR%, downloading it ...
echo Downloading from: %DOWNLOAD_URL%
)
powershell -Command "&{"^
"$webclient = new-object System.Net.WebClient;"^
"if (-not ([string]::IsNullOrEmpty('%MVNW_USERNAME%') -and [string]::IsNullOrEmpty('%MVNW_PASSWORD%'))) {"^
"$webclient.Credentials = new-object System.Net.NetworkCredential('%MVNW_USERNAME%', '%MVNW_PASSWORD%');"^
"}"^
"[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12; $webclient.DownloadFile('%DOWNLOAD_URL%', '%WRAPPER_JAR%')"^
"}"
if "%MVNW_VERBOSE%" == "true" (
echo Finished downloading %WRAPPER_JAR%
)
)
@REM End of extension
@REM Provide a "standardized" way to retrieve the CLI args that will
@REM work with both Windows and non-Windows executions.
set MAVEN_CMD_LINE_ARGS=%*
%MAVEN_JAVA_EXE% ^
%JVM_CONFIG_MAVEN_PROPS% ^
%MAVEN_OPTS% ^
%MAVEN_DEBUG_OPTS% ^
-classpath %WRAPPER_JAR% ^
"-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" ^
%WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
if ERRORLEVEL 1 goto error
goto end
:error
set ERROR_CODE=1
:end
@endlocal & set ERROR_CODE=%ERROR_CODE%
if not "%MAVEN_SKIP_RC%"=="" goto skipRcPost
@REM check for post script, once with legacy .bat ending and once with .cmd ending
if exist "%USERPROFILE%\mavenrc_post.bat" call "%USERPROFILE%\mavenrc_post.bat"
if exist "%USERPROFILE%\mavenrc_post.cmd" call "%USERPROFILE%\mavenrc_post.cmd"
:skipRcPost
@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
if "%MAVEN_BATCH_PAUSE%"=="on" pause
if "%MAVEN_TERMINATE_CMD%"=="on" exit %ERROR_CODE%
cmd /C exit /B %ERROR_CODE%
First_Pipeline/producer/pom.xml 0 → 100644
View file @ be1cb6b5
<?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.0.4</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>producer</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>producer</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>17</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</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>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
First_Pipeline/producer/src/main/java/com/example/producer/ProducerApplication.java 0 → 100644
View file @ be1cb6b5
package com.example.producer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ProducerApplication {
public static void main(String[] args) {
SpringApplication.run(ProducerApplication.class, args);
}
}
First_Pipeline/producer/src/main/java/com/example/producer/controllers/ProducerController.java 0 → 100644
View file @ be1cb6b5
package com.example.producer.controllers;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
@RestController
@RequestMapping("/api/product")
public class ProducerController {
private final KafkaTemplate<String, String> kafkaTemplate;
public ProducerController(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
@GetMapping
public String p(){
for (int i = 0; i < 10; i++) {
kafkaTemplate.send("Test", "SnmpTrap");
//System.out.println("Hello");
}
return "Trap send successfully";
}
}
First_Pipeline/producer/src/main/resources/application.properties 0 → 100644
View file @ be1cb6b5
server.port = 8989
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
First_Pipeline/producer/src/test/java/com/example/producer/ProducerApplicationTests.java 0 → 100644
View file @ be1cb6b5
package com.example.producer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class ProducerApplicationTests {
@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

HIAST - GitLab