package com.timgroup.eventstore.ges.http;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.timgroup.eventstore.api.EventStreamWriter;
import com.timgroup.eventstore.api.NewEvent;
import com.timgroup.eventstore.api.StreamId;
import org.apache.http.HttpHost;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;

import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.UUID;

import static java.util.stream.Collectors.toList;

public class HttpGesEventStreamWriter implements EventStreamWriter {
    private final String host;

    public HttpGesEventStreamWriter(String host) {
        this.host = host;
    }

    @Override
    public void write(StreamId streamId, Collection<NewEvent> events) {
        try {
            CloseableHttpClient client = HttpClientBuilder.create().build();

            HttpPost writeRequest = new HttpPost("/streams/" + streamId.category() + "-" + streamId.id());

            ObjectMapper mapper = new ObjectMapper();

            //todo: non-json body?
            List<HttpEvent> httpEvents = events.stream()
                    .map(e -> {
                        try {
                            return new HttpEvent(
                                    UUID.randomUUID().toString(),
                                    e.type(),
                                    mapper.readTree(e.data()),
                                    mapper.readTree(e.metadata())
                            );
                        } catch (IOException e1) {
                            throw new RuntimeException(e1);
                        }
                    }).collect(toList());

            byte[] bytes = mapper.writer().writeValueAsBytes(httpEvents);

            writeRequest.setEntity(new ByteArrayEntity(bytes, ContentType.create("application/vnd.eventstore.events+json")));

            client.execute(HttpHost.create(host), writeRequest, response -> {
                if (response.getStatusLine().getStatusCode() != 201) {
                    throw new RuntimeException("Write request failed: " + response.getStatusLine());
                }
                return null;
            });
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void write(StreamId streamId, Collection<NewEvent> events, long expectedVersion) {

    }


    private static class HttpEvent {
        public final String eventId;
        public final String eventType;
        public final JsonNode data;
        public final JsonNode metadata;

        private HttpEvent(String eventId, String eventType, JsonNode data, JsonNode metadata) {
            this.eventId = eventId;
            this.eventType = eventType;
            this.data = data;
            this.metadata = metadata;
        }
    }
}

@@ -280,16 +280,82 @@
 reamId;%0A
+import com.timgroup.eventstore.api.WrongExpectedVersionException;%0A
 import o
@@ -729,14 +729,85 @@
 til.
-UUID;%0A
+OptionalLong;%0Aimport java.util.UUID;%0A%0Aimport static java.lang.Long.MIN_VALUE;
 %0Aimp
@@ -923,16 +923,76 @@
 riter %7B%0A
+    private final ObjectMapper mapper = new ObjectMapper();%0A
     priv
@@ -1181,24 +1181,376 @@
 %3E events) %7B%0A
+        write(streamId, events, OptionalLong.empty());%0A    %7D%0A%0A    @Override%0A    public void write(StreamId streamId, Collection%3CNewEvent%3E events, long expectedVersion) %7B%0A        write(streamId, events, OptionalLong.of(expectedVersion));%0A    %7D%0A%0A    private void write(StreamId streamId, Collection%3CNewEvent%3E events, OptionalLong maybeExpectedVersion) %7B%0A
         try 
@@ -1753,85 +1753,170 @@
     
-ObjectMapper mapper = new ObjectMapper();%0A%0A            //todo: non-json body?
+maybeExpectedVersion.ifPresent(expectedVersion -%3E %7B%0A                writeRequest.setHeader(%22ES-ExpectedVersion%22, Long.toString(expectedVersion));%0A            %7D);%0A
 %0A   
@@ -2873,16 +2873,337 @@
 usCode()
+ == 400 && response.getStatusLine().getReasonPhrase().equals(%22Wrong expected EventNumber%22)) %7B%0A                    //todo: remove need for MIN_VALUE%0A                    throw new WrongExpectedVersionException(MIN_VALUE, maybeExpectedVersion.getAsLong());%0A                %7D else if (response.getStatusLine().getStatusCode()
  != 201)
@@ -3467,124 +3467,8 @@
  %7D%0A%0A
-    @Override%0A    public void write(StreamId streamId, Collection%3CNewEvent%3E events, long expectedVersion) %7B%0A%0A    %7D%0A%0A
 %0A   

Add support for optimistic concurrency control