Camel TimerComponent Sınıfı

Örnek

Şöyle yaparız

import org.apache.camel.builder.RouteBuilder;
import org.springframework.stereotype.Component;

@Component
public class RestAPIClientRoute extends RouteBuilder {
  @Override
  public void configure() throws Exception {
    restConfiguration().host("dummy.restapiexample.com").port(80);
    
    from("timer:rest-client?period=10s")
      .to("rest:get:/api/v1/employees")
      .log("${body}");
  }
}

Camel Processors Kavramı

Giriş

Açıklaması şöyle

Processors handles things in between endpoints like routing, transformation, validation, enrichment, etc. 

Örnek

Şu satırı dahil ederiz

<dependency>
  <groupId>org.apache.camel.springboot</groupId>
  <artifactId>camel-activemq-starter</artifactId>
  <version>3.8.0</version>
</dependency>
<dependency>
  <groupId>org.apache.camel.springboot</groupId>
  <artifactId>camel-jackson-starter</artifactId>
  <version>3.8.0</version>
</dependency>

Elimizde ActiveMQ'dan okuyup processor'a gönderen şöyle bir kod olsun. Burada Processor olan GetEmployee .bean() metodu ile belirtiliyor. GetEmployee sınıfının tek metodu olduğu için hangi metodunun çağrılacağını belirtmeye gerek yok.

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.model.dataformat.JsonLibrary;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class ActiveMQReceiverRoute extends RouteBuilder {
  @Autowired
  GetEmployee getEmployee;
  @Override
  public void configure() throws Exception {
    from("activemq:myqueue")
    .unmarshal().json(JsonLibrary.Jackson, Employee.class)
    .bean(getEmployee)
    .to("log:myloggingqueue");
  }
}
@Component
class GetEmployee{
  Logger logger= LoggerFactory.getLogger(GetEmployee.class);

  public void getData(Employee employee){
    logger.info("Emp data: "+employee.getId());
  }
}

Camel Exception Kavramı

Giriş

Exception Route içinde yakalanır Açıklaması şöyle

The clause can either be included inside a specific route definition in order to handle exceptions thrown in said route as shown in the example, or on top of all route definitions in order to catch exceptions that could be thrown anywhere in the router.

It’s important to mention the clause handled(), which determines what to do when an exception is thrown. In this case, since we want to stop the execution and handle the exception, we used handled(true). If we used handled(false), the exception would be rethrown and it could either be caught in a different onException clause or crash the application. The default value is false, which means if we don’t include the clause, the exception will continue to crash the application.

Örnek

Şöyle yaparız. Burada exception bir route içinde yakalanıyor. handled(true) yapılıyor.

from("direct:say-hi")
  .onException(MissingParameterException.class)
    .handled(true)
    .setBody(constant("Expected a query param 'name' that could not be found."))
  .end()
  .choice()
    .when(isNull(header(NAME_TAG)))
      .throwException(new MissingParameterException())
    .otherwise()
      .setProperty(NAME_TAG, header(NAME_TAG))
  .end()
  .setBody(simple(String.format("Hello, ${property.%s}!", NAME_TAG)))
;

Camel Exchange Kavramı

Giriş

Exchange kavramına başlamadan önce Header ve Property kavramını bilmek gerekir.

1. Header

- header("headername") şeklinde erişiriz.

Örnek

"localhost:8080/hello?name=john" şeklindeki istekte name header içinde. header("name") şeklinde erişiriz.

Örnek - header'ın varlığına göre işlem

Şöyle yaparız. Burada when() içinde exchange varlığı kontrol ediliyor. Eğer yoksa otherwise() içinde başka bir işlem yapılıyor

@Component
public class Router extends RouteBuilder {
  private static final String NAME_TAG = "name";
  @Override
  public void configure() throws Exception {
    restConfiguration()
      .bindingMode(RestBindingMode.json);
    rest("hello")
      .get()
      .route()
       .choice()
        .when(isNull(header(NAME_TAG)))
          .setProperty(NAME_TAG, constant("world"))
        .otherwise()
          .setProperty(NAME_TAG, header(NAME_TAG))
        .end()
        .setBody(simple(String.format("Hello, ${property.%s}!", NAME_TAG)));
  }
}

Örnek - header değerine erişme

Eğer header değerinin olduğundan eminsek kontrol etmeksizin şöyle yaparız

@Component
public class Soap2Rest extends RouteBuilder{

  @Override
  public void configure() throws Exception {

    from("direct:celsius-to-fahrenheit")
      .removeHeaders("CamelHttp*")
      .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
          Double.valueOf(exchange.getIn().getHeader("num"));
          ...
        }
      })
      ...
  }
}

2. Property

Açıklaması şöyle

For now, let’s just say properties are values that, unlike headers, live throughout the entire course of the application.

- setProperty("propertyname",propertyvalue) şeklinde değer atarız

- String.format("${property.%s}!", "propertyname") şeklinde değerine erişiriz

3. Exchange

Route'lar arasında dolaşırlar

- Header ve Body alanlarına sahiptir. Açıklaması şöyle

Headers and bodies are volatile, and they are saved in two different Messages or parts of the Exchange: IN (incoming information) and OUT (outgoing information). We call them volatile because they don’t live throughout the entire course of the application, but rather die after reaching an endpoint or processing information from the exchange.

Processor'lar Exchange'lere erişir. 

Örnek - header atama

Şöyle yaparız

@Component
public class MomentOfDayProcessor implements Processor {
  @Override
  public void process(Exchange exchange) throws Exception {
    String momentOfDay = "...";
    exchange.getOut().setHeader("momentOfDay", momentOfDay);
  }
}

Örnek - body atama

Şöyle yaparız

@Component
public class WeatherAdviserProcessor implements Processor {
  @Override
  public void process(Exchange exchange) throws Exception {
    WeatherResponse weatherResponse = exchange.getIn().getBody(WeatherResponse.class);
    ...
    String advice = "...";
    exchange.getOut().setBody(advice);
  }
}

Örnek - body atama

Şöyle yaparız. Burada getOut().setBody() yerine getMessage().setBody() kullanılıyor.

public class HelloWorldProcessor implements Processor {

  @Override
  public void process(Exchange exchange) throws Exception {
    Map<String, Object> result = new HashMap<>();
    result.put("message", "Hello world");
    exchange.getMessage().setBody(result);
  }
}

Açıklaması şöyle

If using InOnly MEP, methods getIn() and getMessage() returns the same instance of Message.

The logic of getMesage() is simple. If exchange have associated out message, return out. Otherwise return in.

In most cases there will be no out message associated with Exchange. In Apache Camel 3 is getOut() deprecated, reserved for edge cases and internal use for component developers. End users are encouraged to prefer getMessage() instead of getIn() and getOut().

Camel RestDefinition Sınıfı

Giriş

1. İsteğe kendimiz cevap verebiliriz.

2. İsteği başka bir bileşene gönderebiliriz.

Örnek - get İsteğine Enpoint İçinde Cevap Verebiliriz

Şöyle yaparız

@Component
public class Router extends RouteBuilder {
  private static final String NAME_TAG = "name";
  @Override
  public void configure() throws Exception {
    restConfiguration()
      .bindingMode(RestBindingMode.json);
    rest("hello")
      .get()
      .route()
       .choice()
        .when(isNull(header(NAME_TAG)))
          .setProperty(NAME_TAG, constant("world"))
        .otherwise()
          .setProperty(NAME_TAG, header(NAME_TAG))
        .end()
        .setBody(simple(String.format("Hello, ${property.%s}!", NAME_TAG)));
  }
}

Örnek - get ile Aynı Sınıftaki Route Tetiklemek

Şöyle yaparız. Bu örnekte istek endpoint'ten from () ile tanımlanan Route'a yönlendiriliyor.

rest()
  .get("hello").route().to("direct:say-hi").endRest()
  .get("bye").route().to("direct:say-bye").endRest()
;
from("direct:say-hi")
  .choice()
    .when(isNull(header(NAME_TAG)))
      .setProperty(NAME_TAG, constant("world"))
    .otherwise()
      .setProperty(NAME_TAG, header(NAME_TAG))
    .end()
    .setBody(simple(String.format("Hello, ${property.%s}!", NAME_TAG)))
;
from("direct:say-bye")
  .setBody(simple("Bye, see you soon!"))
;

Örnek - get ile Başka Sınıftaki Route Tetiklemek

Şöyle yaparız. Bu örnekte istek endpoint'ten from () ile tanımlanan Route'a yönlendiriliyor. Route ise .process(...) aracılığıyla isteği işliyor

@Component
public class RestConfig extends RouteBuilder{

  @Override
  public void configure() throws Exception {
    rest("/convert")

      .get("/celsius/to/fahrenheit/{num}")
        .consumes("text/plain").produces("text/plain")
        .description("Convert a temperature in Celsius to Fahrenheit")
        .param().name("num").type(RestParamType.path)
          .description("Temperature in Celsius")
          .dataType("int").endParam()
        .to("direct:celsius-to-fahrenheit")

      .get("/fahrenheit/to/celsius/{num}")
        .consumes("text/plain").produces("text/plain")
	.description("Convert a temperature in Fahrenheit to Celsius")
	.param().name("num").type(RestParamType.path)
          .description("Temperature in Fahrenheit")
          .dataType("int").endParam()
	.to("direct:fahrenheit-to-celsius");
  }
}

Bu rest noktasını tetiklemek için şöyle yaparız

curl localhost:9090/convert/celcius/to/fahrenheit/50

Parametreye karşı bileşende erişmek için şöyle yaparız

@Component
public class Soap2Rest extends RouteBuilder{

  @Override
  public void configure() throws Exception {

    from("direct:celsius-to-fahrenheit")
      .removeHeaders("CamelHttp*")
      .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
          Double.valueOf(exchange.getIn().getHeader("num"));
          ...
        }
      })
      ...
  }
}

Örnek - get ile Aynı Sınıftaki Route Tetiklemek

Şöyle yaparız. Bu örnekte istek endpoint'ten from () ile tanımlanan Route'a yönlendiriliyor. Route ise .bean(...) metoduyla isteği işliyor

@Component
public class BookRoute extends RouteBuilder {

    private final Environment env;

    public BookRoute(Environment env) {
        this.env = env;
    }

  public void configure() throws Exception {

    restConfiguration()
      .contextPath(env.getProperty("camel.component.servlet.mapping.contextPath", "/rest/*"))
      .apiContextPath("/api-doc")
      .apiProperty("api.title", "Spring Boot Camel Postgres Rest API.")
      .apiProperty("api.version", "1.0")
      .apiProperty("cors", "true")
      .apiContextRouteId("doc-api")
      .port(env.getProperty("server.port", "8080"))
      .bindingMode(RestBindingMode.json);

    rest("/book")
      .consumes(MediaType.APPLICATION_JSON_VALUE)
      .produces(MediaType.APPLICATION_JSON_VALUE)
      .get("/{name}").route()
       .to("{{route.findBookByName}}")
      .endRest()
      .get("/").route()
       .to("{{route.findAllBooks}}")
      .endRest()
      .post("/").route()
       .marshal().json()
       .unmarshal(getJacksonDataFormat(Book.class))
       .to("{{route.saveBook}}")
      .endRest()
      .delete("/{bookId}").route()
        .to("{{route.removeBook}}")
      .end();

      from("{{route.findBookByName}}")
        .log("Received header : ${header.name}")
        .bean(BookService.class, "findBookByName(${header.name})");

    from("{{route.findAllBooks}}")
      .bean(BookService.class, "findAllBooks");


    from("{{route.saveBook}}")
     .log("Received Body ${body}")
     .bean(BookService.class, "addBook(${body})");


    from("{{route.removeBook}}")
      .log("Received header : ${header.bookId}")
     .bean(BookService.class, "removeBook(${header.bookId})");
  }

  private JacksonDataFormat getJacksonDataFormat(Class<?> unmarshalType) {
    JacksonDataFormat format = new JacksonDataFormat();
    format.setUnmarshalType(unmarshalType);
    return format;
  }
}

Örnek - get ile Harici Bir Servise Get Yapmak

Şöyle yaparız. toD() ile bir başka servise çağrı yapıyoruz.

from(CamelConstants.WEATHER_ROUTE)
    .setProperty(CITY_TAG, header(CITY_TAG))
    .removeHeaders("*")
    .setHeader(Exchange.HTTP_METHOD, constant(HttpMethods.GET))
    .toD(generateServiceUrl())
    .unmarshal().json(JsonLibrary.Jackson, WeatherResponse.class)
    .process(weatherAdviserProcessor)

toD() ile kullanılan metod şöyle

private static final String WEATHER_SERVICE_URL =
"http://api.weatherstack.com/current?access_key=%s&query=%s&units=m";

@Value(value = "${api.key}")
private String apiKey;

private String generateServiceUrl() {
  return String.format(WEATHER_SERVICE_URL, apiKey,
    String.format("${property.%s}", CITY_TAG));
}

Gelen cevabı işleyip çeviren processor şöyle

@Component
public class WeatherAdviserProcessor implements Processor {
  @Override
  public void process(Exchange exchange) throws Exception {
    WeatherResponse weatherResponse = exchange.getIn().getBody(WeatherResponse.class);
    ...
    String advice = "...";
    exchange.getOut().setBody(advice);
  }
}

Camel Splitter

Örnek
Şöyle yaparız

from("timer://poll?period=10000").process(new Processor(){
  public void process(Exchange exchange){
    ArrayList<String> list = new ArrayList<>();
    list.add("one");
    list.add("two");
    list.add("three");
    exchange.getIn().setBody(list, ArrayList.class);
  }
})
.split(body())
.log(body().toString())
.to("file:some/dir?fileName=splitted-${id}");

Camel DirectComponent Sınıfı

Giriş
DirectComponent bir çok farklı kanalın birleşme noktası gibi düşünülebilir. Farklı kanallardan gelen girdiler, DirectComponent'e gönderilir. DirectComponent'i dinleyen bir başka kanal da veriyi örneğin bir kuyruğa veya veri tabanına yazar. Açıklaması şöyle

The direct: component provides direct, synchronous invocation of any consumers when a producer sends a message exchange.

Bu bileşen bir thread pool kullanmaz. Açıklaması şöyle

Another difference is Direct component doesn't has any thread pool, the direct consumer process method is invoked by the calling thread of direct producer.

DirecktComponent'e normalde sadece to() ile consumer'lar bağlanır. DirectComponent'i besleyen/tetikleyen kodlar genelde başka yerdedir

Örnek - Log + Consumer

Şöyle yaparız

import org.apache.camel.builder.RouteBuilder;

public class SampleDirectRoute extends RouteBuilder {

  @Override
  public void configure() throws Exception {

    from("direct:sampleInput")
      .log("Received Message is ${body} and headers are ${headers}")
      .to("file:sampleOutput?fileName=output.txt")
    .end();
  }
}

Kullanım Örnekleri

Örnek
- Bir SpringBoot RestController DirectComponent'a select çağrısı gönderir. DirectComponent bir sql cümlesi oluşturur ve JDBCComponent'e gönderir ve cevabı döner.
- SpringBoot RestController DirectComponent'a insert çağrısı gönderir. DirectComponent bir sql cümlesi oluşturur ve JDBCComponent'e gönderir ve cevabı döner.

Örnek
Şöyle yaparız. Activemq'dan okuma yapıp, OrderServer.validate() metoduna gönderir. Bu metod da processOrder kanalına gönderir. Bu kanalı dinleyen OrderService.process() çıktısını Activemq'ya gönderir.

from("activemq:queue:order.in")
    .to("bean:orderServer?method=validate")
    .to("direct:processOrder");

from("direct:processOrder")
    .to("bean:orderService?method=process")
    .to("activemq:queue:order.out");

Örnek
Şöyle yaparız.

// Three endpoints to one "main" route.
from("activemq:queue:order.in")
  .to("direct:processOrder");

from("file:some/file/path")
  .to("direct:processOrder");

from("jetty:http://0.0.0.0/order/in")
  .to("direct:processOrder");

from("direct:processOrder")
  .to("bean:orderService?method=process")
  .to("activemq:queue:order.out");

marshal metodu
Şöyle yaparız

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jackson.JacksonDataFormat;
import org.springframework.stereotype.Component;

import com.javainuse.model.Employee;


@Component
public class RabbitMQRoute extends RouteBuilder {

  @Override
  public void configure() throws Exception {

    JacksonDataFormat jsonDataFormat = new JacksonDataFormat(Employee.class);

    from("direct:startQueuePoint").id("idOfQueueHere").marshal(jsonDataFormat)
    .to("rabbitmq://localhost:5672/javainuse.exchange?queue=javainuse.queue
         &autoDelete=false")
    .end();
  }
}

Camel SftpComponent

Giriş

Açıklaması şöyle

There are multiple ways you can log inside an SFTP server, most widely used 3 types are given below:

User Name and Password (ONE FA)

Public key and private key (ONE FA)

Combination of both 1 and 2 (TWO FA)

flatten Alanı
Şöyle yaparız.

String src="ftp://username:password@host/srcDir/";
String destDir="ftp://username:password@host/destDir/?flatten=true";
fromUri = src+"?recursive=true&delete=true";
        
from(fromUri)
.to(destDir);

Açıklaması şöyle.

SourceDirectory/file1.xml
SourceDirectory/subDir1/file2.xml
SourceDirectory/subDir2/file3.xml
SourceDirectory/subDir3/subDir4/file4.xml

should be moved to a destination Directory

destDir/file1.xml
destDir/file2.xml
destDir/file3.xml
destDir/file4.xml

username Alanı
Şöyle yaparız

@Override
public void configure() throws Exception{
  from("file:/myfolder/")
  .to("sftp://remote_host//mailbox_folder?username=username&" +
      "privateKeyFile=private_key_file")
  .log(LoggingLevel.INFO,"file transferred successfully")
  .end();
}

Camel ProducerTemplate Sınıfı

Giriş
Şu satırı dahil ederiz

import org.apache.camel.ProducerTemplate;

Başka bir Camel bileşenine mesaj göndermek için kullanılır. Camel ve Rest Enpoint'i birleştirmek için iki tane seçenek var

1. Spring RestController tanımlanır. Bu RestController içinden ProducerTemplate ile Camel çağrısı yapılır

2. Camel içinden Rest Endpoint açılır ve Camel kodlarıyla devam edilir.

asyncSendBody metodu
Şöyle yaparız

import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.javainuse.model.Employee;


@RestController
public class SpringRabbitMQController {

  @Produce(uri = "direct:startRabbitMQPoint")
  private ProducerTemplate template;

  @RequestMapping(value = "/employee", method = RequestMethod.GET)
  public String createEmployee(@RequestParam int id, @RequestParam String name,
    @RequestParam String designation) {

    Employee emp = new Employee();
    emp.setName(name);
    emp.setDesignation(designation);
    emp.setEmpId(id);

    template.asyncSendBody(template.getDefaultEndpoint(), emp);
    return "";
  }
}

requestBody metodu

Şöyle yaparız

@RestController
@RequestMapping("/alert/v1")
@CrossOrigin(origins = "*", maxAge = 6000)
public class HelloWorldController {

  @Autowired
  private ProducerTemplate template;

  Gson gson = new Gson();

  @GetMapping(value = "/hello", produces = "application/json")
  public ResponseEntity < String > sayHello() {
    try {
      return ResponseEntity.status(HttpStatus.OK)
        .body(gson.toJson(template.requestBody("direct:sayHello", "Hello from rest 
                 controller")));
    } catch (Exception e) {
      return ResponseEntity.status(HttpStatus.OK).body(null);
    }
  }
}

Camel DefaultCamelContext Sınıfı

Giriş
Şu satırı dahil ederiz

import org.apache.camel.CamelContext;
import org.apache.camel.impl.DefaultCamelContext;

Açıklaması şöyle.

Camel context provides the runtime environment for the whole Apache Camel application. Usually, we will create one context per application and all Camel routes will be executed within the context.

addComponent metodu

AHC WebSocket Component 
Açıklaması şöyle.

The AHC-WS component provides Websocket based endpoints for a client communicating with external servers over Websocket (as a client opening a websocket connection to an external server).

WebSocket Component
Açıklaması şöyle.

Jetty WebSocket Component (and Atmosphere WebSocket component) is intended to expose new WebSocket server. If you need to connect as client to remote WebSocket server, you should use AHC Websocket component

addRoutes metodu
Şöyle yaparız

public static void main(String[] args) {

  RouteBuilder routeBuilder = ...
  CamelContext ctx = new DefaultCamelContext();

  try {
    ctx.addRoutes(routeBuilder);
    ctx.start();
    Thread.sleep(10 * 60 * 1000);
    ctx.stop();
  }
  catch (Exception e) {
    e.printStackTrace();
  }
}

start metodu
Açıklaması şöyle. Bu metod çağrısı non-blocking olduğu için JVM'in exit etmemesine dikkat etmek gerekir.

Camel context implements a Java Service lifecycle interface so it has  start()  and  stop()  methods. In addition to that, it implements  suspend()  and  resume()  methods so it is convenient and easy to manage the context lifecycle. 

Şöyle yaparız.

context.start();

stop metodu
Açıklaması şöyle. Aslında shutdown gibi düşünülebilir.

The operations is paired: start/stop and suspend/resume.

Stop is performing a Graceful shutdown which means all its internal state, cache, etc is cleared. And the routes is being stopped in a graceful manner to ensure messages are given time to complete. If you start a CamelContext after a stop, then its performing a cold start, recreating all the state, cache etc. again.

Şöyle yaparız.

context.stop();

Camel RouteBuilder Sınıfı

Giriş
Şus satırı dahil ederiz

import org.apache.camel.builder.RouteBuilder;

configure metodu
Route kuralları DSL ile yazılır. from ve to ile hangi Component'ten hangi Component'e bilgi akacağı belirtilir. 

Açıklaması şöyle. DSL farklı diller ile yazılabilir. Sanırım en kolayı Fluent API

Apache Camel offers various DSLs, such as Java-based Fluent API, Spring, or Blueprint XML Configuration files, and a Scala DSL. 

- Spring XML DSL is based on the Spring framework and uses XML configuration. 

- Java DSL has the best IDE support.  

- Groovy and Scala DSLs are similar to the Java DSL; in addition, they offer the typical features of modern JVM languages, such as concise code or closures.

Örnek - JMS

Şöyle yaparız

@Component
public class SomeRoute extends RouteBuilder {
  @Override
  public void configure() throws Exception {
    from("jms:{{queue.name}}")
    .process("xmlToJsonProcessor")
    .to("kafka:{{topic}}?brokers={{spring.kafka.bootstrap-servers}}&
      securityProtocol={{spring.kafka.properties.security.protocol}}&
      saslMechanism={{spring.kafka.properties.sasl.mechanism}}&
      saslJaasConfig={{spring.kafka.properties.sasl.jaas.config}}");
  }   
}

Eğer consumer sayısını artırmak istersek şöyle yaparız

@Component
public class SomeRoute extends RouteBuilder {
  @Override
  public void configure() throws Exception {
    from("jms:{{queue.name}}?concurrentConsumers=10")
    .process("xmlToJsonProcessor")
    .to("...");
   }   
}

Örnek - netty
Şöyle yaparız

from("netty:tcp://localhost:9011?textline=true&encoding=ISO-8859-1&sync=true")
.process(new Processor() {
  @Override
  public void process(Exchange exchange) throws InterruptedException {
    String data=exchange.getIn().getBody(String.class);
    ...
  }
})
.to("log:?level=INFO&showBody=true");

Örnek - timer singleshot
Şöyle yaparız

from("timer://runOnce?repeatCount=1&delay=5000")
  .serviceCall("myService");

Örnek - timer periodic
Şöyle yaparız

from("timer://foo?period=100")
.to("direct:bar");

Örnek - timer periodic
Şöyle yaparız

from("timer:theTimer?period=10s")
.log("Timer Invoked . . . ")
.pollEnrich("file:data/input?delete=true&readLock=none")
.log("BODY = [ ${body} ] " )
.to("file:data/output");