@@ -0,0 +1,3160 @@
+package de.openknowledge.jaxrs.reactive.test;%0A%0Aimport java.io.ByteArrayOutputStream;%0Aimport java.io.IOException;%0Aimport java.nio.ByteBuffer;%0Aimport java.nio.channels.AsynchronousFileChannel;%0Aimport java.nio.channels.CompletionHandler;%0Aimport java.nio.file.Files;%0Aimport java.nio.file.Path;%0Aimport java.nio.file.Paths;%0Aimport java.nio.file.StandardOpenOption;%0Aimport java.util.concurrent.Flow.Publisher;%0Aimport java.util.concurrent.Flow.Subscriber;%0Aimport java.util.concurrent.Flow.Subscription;%0A%0Aimport javax.annotation.PostConstruct;%0Aimport javax.enterprise.context.ApplicationScoped;%0Aimport javax.xml.bind.JAXBContext;%0Aimport javax.xml.bind.JAXBException;%0A%0A@ApplicationScoped%0Apublic class CustomerRepository %7B%0A%0A  private JAXBContext context;%0A  private Path path;%0A  %0A  @PostConstruct%0A  public void initialize() throws JAXBException, IOException %7B%0A    context = JAXBContext.newInstance(Customer.class);%0A    path = Paths.get(%22customers.xml%22);%0A    if (!Files.exists(path)) %7B%0A      Files.createFile(path);%0A    %7D%0A  %7D%0A%0A  public void save(Publisher%3CCustomer%3E customers) throws IOException %7B%0A    AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(path, StandardOpenOption.WRITE);%0A%0A    ByteBuffer buffer = ByteBuffer.allocate(1024);%0A    buffer.put(%22%3Ccustomers%3E%22.getBytes());%0A    buffer.flip();%0A    fileChannel.write(buffer, 0, buffer, new CompletionHandler%3CInteger, ByteBuffer%3E() %7B%0A%0A      @Override%0A      public void completed(Integer result, ByteBuffer attachment) %7B%0A        customers.subscribe(new Subscriber%3CCustomer%3E() %7B%0A          %0A          private Subscription subscription;%0A%0A          @Override%0A          public void onNext(Customer customer) %7B%0A            %0A            try %7B%0A              ByteArrayOutputStream outputStream = new ByteArrayOutputStream();%0A              context.createMarshaller().marshal(customer, outputStream);%0A              buffer.put(outputStream.toByteArray());%0A              buffer.flip();%0A              fileChannel.write(buffer, 0, buffer, new CompletionHandler%3CInteger, ByteBuffer%3E() %7B%0A%0A                @Override%0A                public void completed(Integer result, ByteBuffer attachment) %7B%0A                  subscription.request(1);%0A                %7D%0A%0A                @Override%0A                public void failed(Throwable t, ByteBuffer attachment) %7B%0A                  throw new IllegalStateException(t);%0A                %7D%0A              %7D);%0A            %7D catch (JAXBException e) %7B%0A              throw new IllegalStateException(e);%0A            %7D%0A          %7D%0A%0A          @Override%0A          public void onComplete() %7B%0A            buffer.put(%22%3C/customers%3E%22.getBytes());%0A            buffer.flip();%0A            fileChannel.write(buffer, 0);%0A          %7D%0A%0A          @Override%0A          public void onError(Throwable t) %7B%0A            throw new IllegalStateException(t);%0A          %7D%0A%0A          @Override%0A          public void onSubscribe(Subscription subscription) %7B%0A            this.subscription = subscription;%0A            subscription.request(1);%0A          %7D%0A        %7D);%0A      %7D%0A%0A      @Override%0A      public void failed(Throwable t, ByteBuffer attachment) %7B%0A        throw new IllegalStateException(t);%0A      %7D%0A    %7D);%0A  %7D%0A%7D%0A

Add reactive CustomerRepository