public class StdProblem extends Problem {
    
    private StdProblem(ImmutableMap<String, String> problemData) {
        super(problemData);
    }

    public static StdProblem of(WebApplicationException wae) {
        return new StdProblem(parseProblemData(wae.getResponse()));
    }


    public boolean isNotExists() {
        return is(410, "urn:problem:instance-not-exists");
    }

    public static StdProblem newNotExists() {
        return new StdProblem(newProblemdata(410, "urn:problem:instance-not-exists"));
    }

    
    public boolean isMalformedRequestDataProblem() {
        return is(400, "urn:problem:formed-request-data");
    }

    public static StdProblem newMalformedRequestDataProblem() {
        return new StdProblem(newProblemdata(400, "urn:problem:formed-request-data"));
    }
    
    
    public boolean isUnsupportedMimeTypeProblem() {
        return is(415, "urn:problem:unsupported-mimetype");
    }

    public static StdProblem newUnsupportedMimeTypeProblem() {
        return new StdProblem(newProblemdata(415, "urn:problem:unsupported-mimetype"));
    }


    public boolean isUnacceptedMimeTypeProblem() {
        return is(406, "urn:problem:unaccepted-mimetype");
    }

    public static StdProblem newUnacceptedMimeTypeProblem() {
        return new StdProblem(newProblemdata(406, "urn:problem:unaccepted-mimetype"));
    }
    
    
    public boolean isServerErrorProblem() {
        return is(500, "urn:problem:server-error");
    }
   
    public static StdProblem newServerErrorProblem() {
        return new StdProblem(newProblemdata(500, "urn:problem:server-error"));
    }    
} public class Immutables {

    
    /**
     * merges a new entry into a set
     * 
     * @param set         the set 
     * @param entryToAdd  the entry to add
     * @return the new merged immutable set
     */
    public static <T> ImmutableSet<T> join(ImmutableSet<T> set, T entryToAdd) {
        return ImmutableSet.<T>builder().addAll(set).add(entryToAdd).build();
    }


    
    /**
     * merges 2 sets
     * 
     * @param set1  the set1 to merge
     * @param set2  the set2 to merge
     * @return the new merged immutable set
     */
    public static <T> ImmutableSet<T> join(ImmutableSet<T> set1, ImmutableSet<T> set2) {
        return ImmutableSet.<T>builder().addAll(set1).addAll(set2).build();
    }

    
    /**
     * merges a new entry into a list
     * 
     * @param list        the list to merge
     * @param entryToAdd  the entry to add
     * @return the new merged immutable list
     */
    public static <T> ImmutableList<T> join(ImmutableList<T> list, T entryToAdd) {
        return ImmutableList.<T>builder().addAll(list).add(entryToAdd).build();
    }
 
    /**
     * merges a new entry into a list
     * 
     * @param entryToAdd  the entry to add
     * @param list        the list to merge
     * @return the new merged immutable list
     */
    public static <T> ImmutableList<T> join(T entryToAdd, ImmutableList<T> list) {
        return ImmutableList.<T>builder().add(entryToAdd).addAll(list).build();
    }
 
    /**
     * merges 2 lists
     * 
     * @param list1  the list1 to merge
     * @param list2  the list2 to merge
     * @return the new merged immutable list
     */
    public static <T> ImmutableList<T> join(ImmutableList<T> list1, ImmutableList<T> list2) {
        return ImmutableList.<T>builder().addAll(list1).addAll(list2).build();
    }

    
    /**
     * merges a new entry into a map
     * 
     * @param map    the map to merge
     * @param key    the key of the new entry
     * @param value  the value of the new entry
     * @return the new merged immutable map
     */
    public static <K, V> ImmutableMap<K, V> join(ImmutableMap<K, V> map, K key, V value) {
        Map<K, V> m = Maps.newHashMap(map);
        m.put(key, value);
        return ImmutableMap.copyOf(m);
    }

   
    /**
     * merges 2 maps
     * 
     * @param map1   the map1 to merge
     * @param map2   the map2 to merge
     * @return the new merged immutable map
     */
    public static <K, V> ImmutableMap<K, V> join(ImmutableMap<K, V> map1, ImmutableMap<K, V> map2) {
        return ImmutableMap.<K, V>builder().putAll(map1).putAll(map2).build();
    }
    
    
    public static <T> Collector<T, ?, ImmutableList<T>> toList() {
        Supplier<ImmutableList.Builder<T>> supplier = ImmutableList.Builder::new;
        BiConsumer<ImmutableList.Builder<T>, T> accumulator = (b, v) -> b.add(v);
        BinaryOperator<ImmutableList.Builder<T>> combiner = (l, r) -> l.addAll(r.build());
        Function<ImmutableList.Builder<T>, ImmutableList<T>> finisher = ImmutableList.Builder::build;
 
        return Collector.of(supplier, accumulator, combiner, finisher);
    }
 

    public static <T> Collector<T, ?, ImmutableSet<T>> toSet() {
        Supplier<ImmutableSet.Builder<T>> supplier = ImmutableSet.Builder::new;
        BiConsumer<ImmutableSet.Builder<T>, T> accumulator = (b, v) -> b.add(v);
        BinaryOperator<ImmutableSet.Builder<T>> combiner = (l, r) -> l.addAll(r.build());
        Function<ImmutableSet.Builder<T>, ImmutableSet<T>> finisher = ImmutableSet.Builder::build;
 
        return Collector.of(supplier, accumulator, combiner, finisher);
    }
 
    
    public static <T, K, V> Collector<T, ?, ImmutableMap<K, V>> toMap(Function<? super T, ? extends K> keyMapper,
                                                                      Function<? super T, ? extends V> valueMapper) {
        Supplier<ImmutableMap.Builder<K, V>> supplier = ImmutableMap.Builder::new;
        BiConsumer<ImmutableMap.Builder<K, V>, T> accumulator = (b, t) -> b.put(keyMapper.apply(t), valueMapper.apply(t));
        BinaryOperator<ImmutableMap.Builder<K, V>> combiner = (l, r) -> l.putAll(r.build());
        Function<ImmutableMap.Builder<K, V>, ImmutableMap<K, V>> finisher = ImmutableMap.Builder::build;
        
        return Collector.of(supplier, accumulator, combiner, finisher);
    }
}   public class Page {    
    
    private final String path;
    private final ImmutableMap<String, Object> data;
    
    
    
    public Page(String path) {
        this(path, ImmutableMap.of());
    }
    
    protected Page(String path, ImmutableMap<String, Object> data) {
        this.path = path;     
        this.data = data;
    }
    
    public String getPath() {
        return path;
    }
    
    public ImmutableMap<String, Object> getModelMap() {
        return data;
    }
    
    public Page withModelData(String name, Object value) {
        return new Page(path, ImmutableMap.<String, Object>builder().putAll(data).put(name, value).build());
    }
  
    public Page withModelData(ImmutableMap<String, Object> addtionaldata) {
        return new Page(path, ImmutableMap.<String, Object>builder().putAll(data).putAll(addtionaldata).build());
    }

    @Override
    public String toString() {
        return getPath();
    }
} public class LinksBuilder {
        
    private final ImmutableMap<String, Object> links;
    private final URI selfHref;
    
    private LinksBuilder(URI selfHref, ImmutableMap<String, Object> links) {
        this.selfHref = selfHref.toString().endsWith("/") ? selfHref : URI.create(selfHref.toString() + "/");
        this.links = links;
    }

    public static LinksBuilder create(UriInfo uriInfo) {
        return create(uriInfo.getAbsolutePathBuilder().build());
    }
    
    public static LinksBuilder create(URI selfHref) {
        return new LinksBuilder(selfHref, ImmutableMap.of()).withHref("self", selfHref);
    }
    
    public LinksBuilder withHref(String name) {
        return withHref(name, name);
    }
    
    public LinksBuilder withHref(String name, String href) {
        if (name.toLowerCase(Locale.US).startsWith("http")) {
            return withHref(name, URI.create(href));
        } else {
            return withHref(name, URI.create(selfHref.toString() + href));
        }
    }
    
    public LinksBuilder withHref(String name, URI href) {
        return new LinksBuilder(selfHref,
                                ImmutableMap.<String, Object>builder()
                                            .putAll(links)
                                            .put(name, ImmutableMap.of("href", href.toString()))
                                            .build());
    }

   
    public ImmutableMap<String, Object> build() {
        return links;
    }
} public interface ReactiveSource<T> extends Closeable {
    
    CompletableFuture<T> readAsync();

    
    T read();

    
    void consume(Consumer<T> consumer);

    
    void close();
    
    
    static <T> ReactiveSource<T> subscribe(Publisher<T> publisher) {
        return Utils.get(subscribeAsync(publisher)); 
    }
   
    
    static <T> CompletableFuture<ReactiveSource<T>> subscribeAsync(Publisher<T> publisher) {
        CompletableFuture<ReactiveSource<T>> promise = new CompletableFuture<>();
        publisher.subscribe(new ReactiveSourceSubscriber<>(promise));
        
        return promise; 
    }
} public class CompletableKafkaProducer<K, V> extends KafkaProducer<K, V> {
    
    private static final Logger LOG = LoggerFactory.getLogger(CompletableKafkaProducer.class);
    

    public CompletableKafkaProducer(ImmutableMap<String, Object> properties) {
        super(properties);
    }

 
    
    public CompletableFuture<RecordMetadata> sendAsync(ProducerRecord<K, V> record) {
        
        final CompletableFuture<RecordMetadata> promise = new CompletableFuture<>();
        
        
        final Callback callback = new Callback() {
            
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) {
                    LOG.debug("submit into " + metadata.topic() + "  (" + metadata.offset() + "@" + metadata.partition() + ")");
                    promise.complete(metadata);
                } else {
                    LOG.debug(exception.getMessage() + " error occured by writing data");
                    promise.completeExceptionally(exception);
                }
            }
        };

        super.send(record, callback);
        
        
        return promise;
    }
    
} public class KafkaMessage<K, V> {
        
    private final KafkaMessageIdList consumedOffsets;
    private final ConsumerRecord<K, V> record;
    
    public KafkaMessage(KafkaMessageIdList consumedOffsets, ConsumerRecord<K, V> record) {
        this.consumedOffsets = consumedOffsets;
        this.record = record;
    }

    public KafkaMessageIdList getConsumedOffsets() {
        return consumedOffsets;
    }
    
    
    public KafkaMessageId getKafkaMessageId() {
        return KafkaMessageId.valueOf(partition(), offset());
    }

    
    /**
     * The topic this record is received from
     */
    public String topic() {
        return record.topic();
    }

    /**
     * The partition from which this record is received
     */
    public int partition() {
        return record.partition();
    }

    /**
     * The key (or null if no key is specified)
     */
    public K key() {
        return record.key();
    }

    /**
     * The value
     */
    public V value() {
        return record.value();
    }

    /**
     * The position of this record in the corresponding Kafka partition.
     */
    public long offset() {
        return record.offset();
    }
    
    @Override
    public String toString() {
        return consumedOffsets.toString();
    }
} public class KafkaMessageId implements Comparable<KafkaMessageId> {
    private static final boolean IS_BASE64_ENCODED = false;
    
    private final int partition;
    private final long offset;
    
    public KafkaMessageId(int partition, long offset) {
        this.partition = partition;
        this.offset = offset;
    }
    

    public int getPartition() {
        return partition;
    }

    public long getOffset() {
        return offset;
    }
    
    
    @Override
    public int hashCode() {
        return Objects.hash(partition, offset);
    }

    @Override
    public boolean equals(Object other) {
        return (other != null) && 
               (other instanceof KafkaMessageId) && 
               (((KafkaMessageId) other).partition == (this.partition)) &&
               (((KafkaMessageId) other).offset == (this.offset));
    }
    
    @Override
    public int compareTo(KafkaMessageId other) {
        int i = ((Integer) other.partition).compareTo(this.partition);
        if (i == 0) {
            i = ((Long) other.offset).compareTo(this.offset);
        }
        return i;    
    }
    
    
    String asString() {
        return partition + "_" + offset;
    }
    
    @Override
    public String toString() {
        return encode(asString());
    }


    static KafkaMessageId fromString(String id) {
        final int idx = id.indexOf("_");
        return new KafkaMessageId(Integer.parseInt(id.substring(0, idx)), Long.parseLong(id.substring(idx + 1, id.length())));
    }

    
    public static KafkaMessageId valueOf(String id) {
        return fromString(decode(id)); 
    }
    

    public static KafkaMessageId valueOf(int partition, long offset) {
        return new KafkaMessageId(partition, offset);
    }

    

    
    public static Optional<Long> getOffSet(KafkaMessageIdList ids, int partition) {
        
        for (KafkaMessageId id : ids) {
            if (id.getPartition() == partition) {
                return Optional.of(id.getOffset());
            }
        }
        
        return Optional.empty();
    }

   
    static String decode(String s) {
        return IS_BASE64_ENCODED ? new String(Base64.getUrlDecoder().decode(s), Charsets.UTF_8) : s;
    }

    static String encode(String s) {
        return  IS_BASE64_ENCODED ? Base64.getUrlEncoder().encodeToString(s.getBytes(Charsets.UTF_8)) : s;
    }
} public class KafkaMessageIdList implements Iterable<KafkaMessageId> {

    private final ImmutableList<KafkaMessageId> ids;
    
    
    public KafkaMessageIdList() {
        this(ImmutableList.of());
    }

    private KafkaMessageIdList(ImmutableList<KafkaMessageId> ids) {
        this.ids = ids;
    }

    public boolean isEmpty() {
        return ids.isEmpty();
    }
    
    public int size() {
        return ids.size();
    }
    
    public KafkaMessageIdList merge(KafkaMessageIdList other) {
        return new KafkaMessageIdList(ImmutableList.<KafkaMessageId>builder().addAll(this.ids).addAll(other.ids).build());
    }


    public KafkaMessageIdList erase(KafkaMessageId id) {
        List<KafkaMessageId> idList = Lists.newArrayList(ids);
        idList.remove(id);
        return new KafkaMessageIdList(ImmutableList.copyOf(idList));
    }

    
  
    public KafkaMessageIdList replacePartitionOffset(int partition, long offset) {
        List<KafkaMessageId> newIdentifiers = Lists.newArrayList(ids);
        
        // remove if partition offset already exists
        for (KafkaMessageId id : ImmutableList.copyOf(ids)) {
            if (id.getPartition() == partition) {
                newIdentifiers.remove(id);
            }
        }

        // add new partition offset
        newIdentifiers.add(KafkaMessageId.valueOf(partition, offset));
        
        return new KafkaMessageIdList(ImmutableList.copyOf(newIdentifiers));
    }

    
    
    public boolean contains(KafkaMessageId id) {
        return ids.contains(id);
    }
    
    @Override
    public Iterator<KafkaMessageId> iterator() {
        return ids.iterator();
    }
    
    @Override
    public String toString() {
        return KafkaMessageId.encode(Joiner.on(",")
                                           .join(ids.stream()
                                                    .map(id -> id.asString())
                                                    .collect(Collectors.toList())));
    }

    
    public static KafkaMessageIdList of(KafkaMessageId... ids) {
        return new KafkaMessageIdList(ImmutableList.copyOf(ids));
    }
  
    
    public static KafkaMessageIdList valueOf(String ids) {
        ImmutableList<KafkaMessageId> idList;
        if (Strings.isNullOrEmpty(ids)) {
            idList = ImmutableList.of();
        } else {
            idList = ImmutableList.copyOf(Splitter.on(",")
                                                  .trimResults()    
                                                  .splitToList(KafkaMessageId.decode(ids))
                                                  .stream()
                                                  .map(id -> KafkaMessageId.fromString(id))
                                                  .collect(Collectors.toList()));
        }
        
        return new KafkaMessageIdList(idList);
    }

    
} public class KafkaSource<K, V> {
    
    private final ImmutableMap<String, Object> properties;
    
    
    public KafkaSource(ImmutableMap<String, Object> properties) {
        this.properties = properties;
    }
 
    
    
    public SingleKafkaSource<K, V> withTopic(String topic) {
        return new SingleKafkaSourceImpl<K, V>(this.properties, topic, new KafkaMessageIdList(), new KafkaMessageIdList());
    }
    
    
    public ImmutableSet<String> listTopics() {
        try (KafkaConsumer<Object, Object> consumer = new KafkaConsumer<>(ImmutableMap.copyOf(properties))) {
            return ImmutableSet.copyOf(consumer.listTopics().keySet());
        }        
    }
    



    public static interface SingleKafkaSource<K, V> extends Publisher<KafkaMessage<K, V>> {
                
        /**
         * @return the new source instance
         * @throws ConnectException if an connect error occurs
         */
        default ReactiveSource<KafkaMessage<K, V>> open() throws ConnectException {
            return Utils.get(openAsync());
        } 

        /**
         * @return the new source instance future
         */
        default CompletableFuture<ReactiveSource<KafkaMessage<K, V>>> openAsync() {
            return ReactiveSource.subscribeAsync(this);
        }
        
        
        SingleKafkaSource<K, V> fromOffsets(KafkaMessageIdList consumedOffsets); 
        
        
        SingleKafkaSource<K, V> filter(KafkaMessageIdList messageIds);
    }
    
    
    
    
    private static class SingleKafkaSourceImpl<K, V> implements SingleKafkaSource<K, V> {
        
        private final ImmutableMap<String, Object> properties;
        private final String topic; 
        private final KafkaMessageIdList consumedOffsets;
        private final KafkaMessageIdList idsToFilter;



        private SingleKafkaSourceImpl(ImmutableMap<String, Object> properties, 
                                      String topic, 
                                      KafkaMessageIdList consumedOffsets,
                                      KafkaMessageIdList idsToFilter) {
            this.properties = properties;
            this.topic = topic;
            this.consumedOffsets = consumedOffsets;
            this.idsToFilter = idsToFilter;
        }

        
        
        @Override
        public SingleKafkaSourceImpl<K, V> fromOffsets(KafkaMessageIdList consumedOffsets) {
            return new SingleKafkaSourceImpl<>(this.properties,         
                                               this.topic, 
                                               consumedOffsets,
                                               this.idsToFilter);
        }

        
        @Override
        public SingleKafkaSourceImpl<K, V> filter(KafkaMessageIdList idsToFilter) {
            return new SingleKafkaSourceImpl<>(this.properties,         
                                               this.topic, 
                                               this.consumedOffsets,
                                               idsToFilter);
        }
        
        
        @Override
        public void subscribe(Subscriber<? super KafkaMessage<K, V>> subscriber) {
            // https://github.com/reactive-streams/reactive-streams-jvm#1.9
            if (subscriber == null) {  
                throw new NullPointerException("subscriber is null");
            }
            
            new ConsumerSubscription<K, V>(topic, properties, consumedOffsets, idsToFilter, subscriber);
        }
        
        
        
        private static class ConsumerSubscription<K, V> implements Subscription {
            
            private final SubscriberNotifier<KafkaMessage<K, V>> subscriberNotifier; 
            private final AtomicBoolean isOpen = new AtomicBoolean(true);
            private final InboundBuffer inboundBuffer;

            
            private ConsumerSubscription(String topic, 
                                         ImmutableMap<String, Object> properties, 
                                         KafkaMessageIdList consumedOffsets,
                                         KafkaMessageIdList idsToFilter,
                                         Subscriber<? super KafkaMessage<K, V>> subscriber) {

                this.subscriberNotifier = new SubscriberNotifier<>(subscriber, this);

                
                KafkaConsumer<K, V> consumer = new KafkaConsumer<>(ImmutableMap.copyOf(properties));
            
            
                
                // seek to offset
                ImmutableList<TopicPartition> partitions = getTopicPartitions(topic, consumer);
                consumer.assign(partitions);
            
                for (TopicPartition partition : partitions) {
                    Optional<Long> offset = KafkaMessageId.getOffSet(consumedOffsets, partition.partition());
                    if (offset.isPresent()) {
                        consumer.seek(partition, offset.get() + 1);
                    } else {
                        consumer.seekToBeginning(partition);
                    }
                }
                 
                
                this.inboundBuffer = new InboundBuffer(consumedOffsets,
                                                       idsToFilter,
                                                       consumer, 
                                                       (record) -> subscriberNotifier.notifyOnNext(record));
                Thread t = new Thread(inboundBuffer);
                t.setDaemon(true);
                t.start(); 
                
                subscriberNotifier.start();
            }
            
            
            private static <K, V> ImmutableList<TopicPartition> getTopicPartitions(String topic, KafkaConsumer<K, V> consumer) {
                List<TopicPartition> partitions =  consumer.partitionsFor(topic)
                                                           .stream()
                                                           .map(partition -> new TopicPartition(topic, partition.partition()))
                                                           .collect(Collectors.toList());
                return ImmutableList.copyOf(partitions);
            }
            
                    
            
            @Override
            public void cancel() {
                subscriberNotifier.notifyOnComplete();
            } 
            
            
            @Override
            public void request(long n) {
                if (isOpen.get()) {
                    if(n < 0) {
                        // https://github.com/reactive-streams/reactive-streams#3.9
                        subscriberNotifier.notifyOnError((new IllegalArgumentException("Non-negative number of elements must be requested: https://github.com/reactive-streams/reactive-streams#3.9")));
                        
                    } else {
                        inboundBuffer.onRequested((int) n);
                    }
                } else {
                    subscriberNotifier.notifyOnError((new IllegalArgumentException("source is closed")));
                }
            }


             
            @Override
            public String toString() {
               return "";
            }
            

            private class InboundBuffer implements Runnable {
                private final Queue<KafkaMessage<K, V>> bufferedRecords = Queues.newConcurrentLinkedQueue();
                private final Consumer<KafkaMessage<K, V>> recordConsumer;
                
                private final KafkaConsumer<K, V> consumer;
                private final AtomicInteger numPendingRequested = new AtomicInteger(0);
                
                private final Object processingLock = new Object();
                private KafkaMessageIdList consumedOffsets;
                private Optional<KafkaMessageIdList> optionalIdsToFilter;

                
                
                public InboundBuffer(KafkaMessageIdList consumedOffsets,
                                     KafkaMessageIdList idsToFilter,
                                     KafkaConsumer<K, V> consumer, 
                                     Consumer<KafkaMessage<K, V>> recordConsumer) {
                    
                    this.consumedOffsets = consumedOffsets;
                    this.optionalIdsToFilter = idsToFilter.isEmpty() ? Optional.empty() : Optional.of(idsToFilter);
                    this.consumer = consumer;
                    this.recordConsumer = recordConsumer;
                }

                public void onRequested(int num) {
                    numPendingRequested.addAndGet(num);
                    process();
                }
                

                public void onRecords(ConsumerRecords<K, V> records) {
                    
                    for (ConsumerRecord<K, V> record : records) {
                        
                        if (optionalIdsToFilter.isPresent()) {
                            KafkaMessageId msgId = KafkaMessageId.valueOf(record.partition(), record.offset());
                            if (optionalIdsToFilter.get().contains(msgId)) {
                                optionalIdsToFilter = Optional.of(optionalIdsToFilter.get().erase(msgId));
                            } else {
                                continue;
                            }
                        }

                        KafkaMessageIdList newConsumedOffsets = consumedOffsets.replacePartitionOffset(record.partition(), record.offset());
                        this.consumedOffsets = newConsumedOffsets;
                        
                        KafkaMessage<K, V> msg = new KafkaMessage<>(newConsumedOffsets, record);
                        bufferedRecords.add(msg);
                    }
                    
                    process();
                }
                
                
                private void process() {
                    
                    while (numPendingRequested.get() > 0) {
                        
                        synchronized (processingLock) {
                            KafkaMessage<K, V> record = bufferedRecords.poll();
                            if (record == null) {
                                
                                if (optionalIdsToFilter.isPresent()) {
                                    if (optionalIdsToFilter.get().isEmpty()) {
                                        cancel();
                                    }
                                }
                                
                                return;
                            } else {
                                recordConsumer.accept(record);
                                numPendingRequested.decrementAndGet();
                            }
                        }
                    }
                }
                 
                
                @Override
                public void run() {
                    while (true) {
                        onRecords(consumer.poll(100));
                    }
                }
            }
        }
    }  
}   public class AvroMessage {
    private static final Logger LOG = LoggerFactory.getLogger(AvroMessage.class);
    

    private final GenericRecord avroMessage;
    
    private AvroMessage(GenericRecord avroMessage) {
        this.avroMessage = avroMessage;
    }
    
    
    public GenericRecord getGenericRecord() {
        return avroMessage;
    }

    public Schema getSchema() {
        return avroMessage.getSchema();
    }
    
    public MediaType getMimeType() {
        return MediaType.valueOf("application/vnd." + Joiner.on(".").join(getSchema().getNamespace(), getSchema().getName().replace("_v", "-v")) + "+json");
    }
    
    public byte[] getBinaryData() throws AvroSerializationException {
        return serialize(avroMessage);
    }
    
    @Override
    public String toString() {
        return avroMessage.toString();
    }
    
    static AvroMessage from(GenericRecord avroRecord) {
        return new AvroMessage(avroRecord);
    }
    
    
    static AvroMessage from(byte[] serialized, ImmutableMap<String, Schema> schemaRegistry, ImmutableList<Schema> readerSchemas) {
        ByteBuffer buffer = ByteBuffer.wrap(serialized);
        int headerLength = buffer.getInt();
        
        byte[] headerBytes = new byte[headerLength];
        buffer.get(headerBytes);
        ImmutableList<String> namespaceName = ImmutableList.copyOf(Splitter.on('#').splitToList(new String(headerBytes, Charsets.UTF_8)));
        Schema writerSchema = schemaRegistry.get(namespaceName.get(0) + "." + namespaceName.get(1));
        
        byte[] msg = new byte[buffer.remaining()];
        buffer.get(msg);
        
        return new AvroMessage(deserializeAvroMessage(msg, writerSchema, readerSchemas));
    }
    
    
    private static GenericRecord deserializeAvroMessage(byte[] bytes, Schema writerSchema, ImmutableList<Schema> readerSchemas) {
        
        for (Schema readerSchema : readerSchemas) {
            try {
                return deserializeAvroMessage(bytes, writerSchema, readerSchema);
            } catch (RuntimeException ignore) { }
        }
        
        return deserializeAvroMessage(bytes, writerSchema, (Schema) null);
    }


    private static GenericRecord deserializeAvroMessage(byte[] bytes, Schema writerSchema, Schema readerSchema) {
        try {
            DatumReader<GenericRecord> reader = (readerSchema == null) ? new SpecificDatumReader<GenericRecord>(writerSchema)
                                                                       : new SpecificDatumReader<GenericRecord>(writerSchema, readerSchema) ;
            Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
            GenericRecord avroRecord = reader.read(null, decoder);
            return avroRecord;
        } catch (IOException ioe) {
            LOG.debug("error occured by deserialize avro record", ioe);
            throw new RuntimeException(ioe);
        } catch (AvroTypeException ae) {
            LOG.debug("error occured by  deserialize avro record", ae);
            throw ae;
        }
    }
    
    
    
    
    private static byte[] serialize(GenericRecord avroMessage) throws AvroSerializationException {
        try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
            final Encoder encoder = EncoderFactory.get().binaryEncoder(os, null); 

            final GenericDatumWriter<GenericRecord> writer = new GenericDatumWriter<GenericRecord>(avroMessage.getSchema()); 
            writer.write(avroMessage, encoder); 
            encoder.flush(); 
            
            return withHeader(avroMessage.getSchema(), os.toByteArray()); 
        } catch(IOException | RuntimeException rt) {
            throw new AvroSerializationException("serializing avro message failed (wrong schema type?) " + rt.getMessage(), avroMessage.getSchema().getFullName(), rt);
        }
    }
    
    
    private static byte[] withHeader(Schema schema, byte[] msg) {
        final byte[] header = (schema.getNamespace() + "#" + schema.getName()).getBytes(Charsets.UTF_8);
        final byte[] lengthHeader = ByteBuffer.allocate(4).putInt(header.length).array();
         
        final byte[] newBytes = new byte[4 + header.length + msg.length];
        System.arraycopy(lengthHeader, 0, newBytes, 0, lengthHeader.length);
        System.arraycopy(header, 0, newBytes, lengthHeader.length, header.length);
        System.arraycopy(msg, 0, newBytes, 4 + header.length, msg.length);
        
        return newBytes;
    }    
} public class AvroMessageMapperRepository implements Closeable {
    
    private static final Logger LOG = LoggerFactory.getLogger(AvroMessageMapperRepository.class);

    private final ReplicationJob replicationJob; 
    private volatile ImmutableList<SchemaInfo> erroneousSchemas = ImmutableList.of();
    private volatile ImmutableMap<String, AvroMessageMapper> mapper = ImmutableMap.of();
    private volatile ImmutableMap<String, String> schemaNameIndex = ImmutableMap.of();
    private volatile ImmutableMap<String, Schema> schemaIndex = ImmutableMap.of();

    
    
    
    public AvroMessageMapperRepository(final URI uri) {
        this.replicationJob = ReplicationJob.source(uri)
                                            .withRefreshPeriod(Duration.ofMinutes(1))
                                            .startConsumingBinary(this::onRefresh);
    }
 
    
    public void onRefresh(final byte[] binary) {
        ImmutableSet<Pair<AvroMessageMapper, SchemaInfo>> results = parseSchemas(binary).stream()
                                                                                        .map(schema -> AvroMessageMapper.createrMapper(schema))
                                                                                        .collect(Immutables.toSet());

        
        erroneousSchemas = results.stream()
                                  .filter(result -> (result.getFirst() == null))
                                  .map(result -> result.getSecond())
                                  .collect(Immutables.toList());

        
        this.mapper = results.stream()
                             .filter(result -> (result.getFirst() != null))
                             .map(result -> result.getFirst())
                             .collect(Immutables.toMap(AvroMessageMapper::getMimeType,
                                      (AvroMessageMapper m) -> m));
        
        this.schemaNameIndex = mapper.values()
                                     .stream()
                                     .collect(Immutables.toMap(m -> m.getSchema().getFullName(), 
                                                               m -> m.getMimeType()));
        
        this.schemaIndex = mapper.values()
                                 .stream()
                                 .collect(Immutables.toMap(m -> m.getSchema().getFullName(), 
                                                           m -> m.getSchema()));
    }    
    
    
 
    
    @Override
    public void close() throws IOException {
        replicationJob.close();
    }


    public ImmutableMap<String, String> getRegisteredSchemasAsText() {
        final Map<String, String> schemas = Maps.newHashMap();
        
        for (Entry<String, AvroMessageMapper> entry : mapper.entrySet()) {
            schemas.put(entry.getKey(), entry.getValue().toString());
            schemas.put(entry.getKey().replace("+json", ".list+json"), "[" + entry.getValue().toString() + "\r\n]");
        }
        
        return ImmutableMap.copyOf(schemas);
    }
    

    public  ImmutableMap<String, Schema> getRegisteredSchemas() {
        return mapper.entrySet().stream().collect(Immutables.toMap(e -> e.getKey(), e -> e.getValue().getSchema()));
    }
    
    
    public ImmutableList<SchemaInfo> getErroneousSchemas() {
        return erroneousSchemas;
    }

    
    public Optional<String> getRegisteredSchema(final MediaType mimetype) {
        for (Entry<String, String> entry : getRegisteredSchemasAsText().entrySet()) {
            if (entry.getKey().equals(mimetype.toString())) {
                return Optional.of(entry.getValue().toString());
            }
        }
        
        return Optional.empty();
    }
    

    public JsonObject toJson(final AvroMessage avroMessage) {
        return getJsonToAvroMapper(avroMessage.getSchema()).map(mapper -> mapper.toJson(avroMessage))
                                                           .orElseThrow(() -> new SchemaException("unsupported type", avroMessage.getMimeType().toString()));
    }

    
    public ImmutableList<AvroMessage> toAvroMessages(final InputStream jsonObjectStream, final String mimeType) {
        return getJsonToAvroMapper(mimeType).map(mapper -> mapper.toAvroMessages(Json.createParser(jsonObjectStream)))
                                            .orElseThrow(() -> new SchemaException("unsupported type", mimeType));
    }
    

    public AvroMessage toAvroMessage(final InputStream jsonObjectStream, final String mimeType) {
        return getJsonToAvroMapper(mimeType).map(mapper -> mapper.toAvroMessage(Json.createParser(jsonObjectStream)))
                                            .orElseThrow(() -> new SchemaException("unsupported type", mimeType));
    }
    

    public AvroMessage toAvroMessage(final byte[] serializedAvroMessage, final MediaType readerMimeType) throws SchemaException {
        return toAvroMessage(serializedAvroMessage, ImmutableList.of(readerMimeType));
    }
    

    public AvroMessage toAvroMessage(final byte[] serializedAvroMessage, final ImmutableList<MediaType> readerMimeTypes) throws SchemaException {
        final ImmutableList<Schema> readerSchemas = readerMimeTypes.stream()
                                                                   .map(mimeType -> getJsonToAvroMapper(mimeType.toString()))
                                                                   .filter(optionalMapper -> optionalMapper.isPresent())
                                                                   .map(mapper -> mapper.get().getSchema())
                                                                   .collect(Immutables.toList());
        
        final AvroMessage avroMessage = AvroMessage.from(serializedAvroMessage, schemaIndex, readerSchemas);

        for (MediaType readerMimeType : readerMimeTypes) {
            if (avroMessage.getMimeType().isCompatible(readerMimeType)) {
                return avroMessage;
            }
        }

        throw new SchemaException("avro messaqge type " + avroMessage.getMimeType() + " does not match with a requested type " + readerMimeTypes);
    }
    

    public ServerSentEvent toServerSentEvent(final KafkaMessageIdList consumedOffsets, final AvroMessage avroMessage) {
        return getJsonToAvroMapper(avroMessage.getSchema()).map(schema -> ServerSentEvent.newEvent()
                                                                                         .id(consumedOffsets.toString())
                                                                                         .data(avroMessage.getMimeType().toString() + "\n" + toJson(avroMessage).toString()))
                                                           .orElseThrow(() -> new SchemaException("unsupported type", avroMessage.getMimeType().toString()));
    }
    
    
    Schema getSchema(final String namespace, final String name) {
        return getJsonToAvroMapper(namespace, name).orElseThrow(() -> new SchemaException("unsupported type", namespace + "." + name))
                                                   .getSchema();
    }
    
    
    private Optional<AvroMessageMapper> getJsonToAvroMapper(final Schema schema) {
        return getJsonToAvroMapper(schema.getNamespace(), schema.getName());
    }
    

    private Optional<AvroMessageMapper> getJsonToAvroMapper(final String namespace, final String name) {
        return  Optional.ofNullable(schemaNameIndex.get(namespace + "." + name))
                        .map(mimeType -> mapper.get(mimeType));
    }

    
    private Optional<AvroMessageMapper> getJsonToAvroMapper(final String mimeType) {
        return Optional.ofNullable(mapper.get(mimeType));
    }
    
   
    

    
    private static ImmutableList<SchemaInfo> parseSchemas(final byte[] binary) {
        try {
            // unpack within temp dir 
            final Path tempDir = Files.createTempDirectory("avro_repo_" + new Random().nextInt(Integer.MAX_VALUE) + "_");
            Zip.unpack(binary, tempDir);

            
            // scan avro scheme files
            final AvroSchemeFileVisitor avroSchemeFileVisitor = new AvroSchemeFileVisitor();
            Files.walkFileTree(tempDir, avroSchemeFileVisitor);

            // scan avro idl files
            final AvroIdlFileVisitor avroIdlFileVisitor = new AvroIdlFileVisitor();
            Files.walkFileTree(tempDir, avroIdlFileVisitor);

            
            // return schema sets
            return Immutables.join(avroSchemeFileVisitor.getCollectedSchemas(),
                                   avroIdlFileVisitor.getCollectedSchemas())
                             .stream()
                             .distinct()
                             .collect(Immutables.toList());
            
            
            
        } catch (IOException ioe) {
            LOG.warn("error loading avro schema", ioe);
            return ImmutableList.of();
        }
    }
    
    
    
    private static final class AvroSchemeFileVisitor extends SimpleFileVisitor<Path> {
        private final List<SchemaInfo> schemas = Lists.newArrayList();

        public ImmutableList<SchemaInfo> getCollectedSchemas() {
            return ImmutableList.copyOf(schemas);
        }
        
        @Override
        public FileVisitResult visitFile(final Path path, final BasicFileAttributes attrs) throws IOException {
            
            if (path.getFileName().toString().endsWith(".avsc")) {
                
            //    try (InputStream is = new FileInputStream(path.toFile())) {
                    schemas.add(new SchemaInfo(path.toFile().getAbsolutePath(), 
                                attrs.lastModifiedTime().toInstant(), 
                                new Schema.Parser().parse(path.toFile())));
            //    }
            }
                
            return FileVisitResult.CONTINUE;
        }
    }
    
    
    private static final class AvroIdlFileVisitor extends SimpleFileVisitor<Path> {
        private final List<SchemaInfo> schemas = Lists.newArrayList();

        public ImmutableList<SchemaInfo> getCollectedSchemas() {
            return ImmutableList.copyOf(schemas);
        }
        
        @Override
        public FileVisitResult visitFile(final Path path, final BasicFileAttributes attrs) throws IOException {
            
            if (path.getFileName().toString().endsWith(".avdl")) {
                
                //try (InputStream is = new FileInputStream(path.toFile())) {
    
                try {
                    //for (Schema schema : idlToJsonSchemas(is)) {
                    for (Schema schema : idlToJsonSchemas(path.toFile())) {
                        schemas.add(new SchemaInfo(path.toFile().getAbsolutePath(), 
                                    attrs.lastModifiedTime().toInstant(), 
                                    schema));
                    }
                } catch (ParseException | IOException e) {
                    throw new RuntimeException("error occured by parsing " + path.toFile().getAbsolutePath(), e);
                }
            }
                
            return FileVisitResult.CONTINUE;
        }
        
        
        private ImmutableSet<Schema> idlToJsonSchemas(final File file) throws IOException, ParseException {
            try (Idl parser = new Idl(file)) {
                return parser.CompilationUnit()
                             .getTypes()
                             .stream()
                             .filter(scheme -> (scheme.getType() == Type.RECORD))
                             .collect(Immutables.toSet());

            } 
        }            
    }
    
 
    
    private static class Zip {

        private Zip() { }
        
        public static void unpack(final byte[] binary, final Path targetDir) throws IOException {
            final ZipInputStream zin = new ZipInputStream(new ByteArrayInputStream(binary));

            ZipEntry ze = null;
            while ((ze = zin.getNextEntry()) != null) {
                final File file = new File(targetDir + File.separator + ze.getName());
                if (ze.isDirectory()) {
                    file.mkdir();
                } else {
                    file.toPath().getParent().toFile().mkdirs();
                    Files.write(file.toPath(), ByteStreams.toByteArray(zin));
                }
                zin.closeEntry();
            }
            zin.close();
        }
    }
} public class ResultConsumer implements BiConsumer<Object, Throwable> {
    
    private final AsyncResponse asyncResponse;
    
    private ResultConsumer(AsyncResponse asyncResponse) {
        this.asyncResponse = asyncResponse;
    }

    
    @Override
    public void accept(Object result, Throwable error) {
        if (error == null) {
            asyncResponse.resume(result);
        } else {
            asyncResponse.resume(Utils.unwrap(error));
        }
    }
    
    
    /**
     * forwards the response to the REST response object. Includes error handling also 
     * @param asyncResponse the REST response
     * @return the BiConsumer consuming the response/error pair
     */
    public static final BiConsumer<Object, Throwable> writeTo(AsyncResponse asyncResponse) {
        return new ResultConsumer(asyncResponse);
    }
} public interface ServerSentEvent {
    
    
    /**
     * @return the comment or NULL
     */
    Optional<String> getComment();
    
    /**
     * @return the data ('data:' value)  or NULL
     */
    Optional<String> getData();
    
    
    /**
     * @return the event type ('event:' value) or NULL
     */
    Optional<String> getEvent();
  
    
    /**
     * @return the id ('id:' value) or NULL
     */
    Optional<String> getId();

    
    /**
     * @return the retry ('retry:' value) or NULL
     */
    Optional<Integer> getRetry();
    
    
    default boolean isSystem() {
        return !(getData().isPresent() || getEvent().isPresent());
    }


    /**
     * @return the "wire" formated event according to Server-Sent Events spec
     */
    @Override
    String toString();
  
    
    /**
     * @return the wire formatted sse 
     */
    default String toWire() {
        StringBuilder sb = new StringBuilder();
        
        getComment().ifPresent(comment -> sb.append(": " + comment + "\r\n"));
        getId().ifPresent(id -> sb.append("id: " + id + "\r\n"));
        getEvent().ifPresent(event -> sb.append("event: " + event + "\r\n"));
        getData().ifPresent(data -> Splitter.on("\n").splitToList(data).forEach(line -> sb.append("data: " + line + "\r\n")));
        getRetry().ifPresent(retry -> sb.append("retry: " + retry + "\r\n"));
        sb.append("\r\n");
        
        return sb.toString();
    }
    
    
    
    /**
     * @return new event
     */
    public static SSEBuildableEvent newEvent() {
        return new SSEBuildableEvent(null, null, null, null, null);
    }
    
    
    
    
    /**
     * A buildable sse event
     */
    public static class SSEBuildableEvent implements ServerSentEvent {
        private final String id;
        private final String event;
        private final String data;
        private final String comment;
        private final Integer retry;


        private SSEBuildableEvent(String id, String event, String data, String comment, Integer retry) {
            this.id = id;
            this.event = event;
            this.data = data;
            this.comment = comment;
            this.retry = retry;
        }
        
        
        /**
         * @param data  the data to set 
         * @return the "cloned", updated event
         */
        public SSEBuildableEvent data(String data) {
            return new SSEBuildableEvent(this.id,
                                         this.event, 
                                         data, 
                                         this.comment, 
                                         this.retry); 
        }
        
        /**
         * @param id  the is to set
         * @return the "cloned", updated event
         */
        public SSEBuildableEvent id(String id) {
            return new SSEBuildableEvent(id, 
                                         this.event, 
                                         this.data, 
                                         this.comment, 
                                         this.retry); 
        }
        
        /**
         * @param comment  the comment to set
         * @return the "cloned", updated event
         */
        public SSEBuildableEvent comment(String comment) {
            return new SSEBuildableEvent(this.id, 
                                         this.event, 
                                         this.data, 
                                         comment, 
                                         this.retry); 
        }
        
        /**
         * @param event  the event to set
         * @return the "cloned", updated event
         */
        public SSEBuildableEvent event(String event) {
            return new SSEBuildableEvent(this.id, 
                                         event, 
                                         this.data, 
                                         this.comment,
                                         this.retry); 
        }
        
        /**
         * @param retry the retry time to set
         * @return the "cloned", updated event
         */
        public SSEBuildableEvent retry(Integer retry) {
            return new SSEBuildableEvent(this.id, 
                                         this.event, 
                                         this.data, 
                                         this.comment,  
                                         retry); 
        }

        @Override
        public Optional<String> getComment() {
            return Optional.ofNullable(comment);
        }
        
        @Override
        public Optional<String> getData() {
            return Optional.ofNullable(data);
        }

        @Override
        public Optional<String> getEvent() {
            return Optional.ofNullable(event);
        }

        @Override
        public Optional<String> getId() {
            return Optional.ofNullable(id);
        }

        @Override
        public Optional<Integer> getRetry() {
            return Optional.ofNullable(retry);
        }
        
        /**
         * @return if event contains data or event info
         */
        boolean isUserEvent() {
            return (getData() != null) || (getEvent() != null);
        }
        
        
        @Override
        public String toString() {
            return toWire();
        }    
        
        
        @Override
        public boolean equals(Object other) {
            return (other instanceof ServerSentEvent) ? ((ServerSentEvent) other).toWire().equals(this.toWire()) : false;
        }
        
        @Override
        public int hashCode() {
           return toWire().hashCode();
        }
    }
} public class ServletSseSubscriber implements Subscriber<ServerSentEvent> {
    private static final Logger LOG = LoggerFactory.getLogger(ServletSseSubscriber.class);

    
    private final AtomicBoolean isOpen = new AtomicBoolean(true);
    private final AtomicReference<Subscription> subscriptionRef = new AtomicReference<>(new IllegalStateSubscription());
    
    private final SseOutboundChannel channel;

    

    public ServletSseSubscriber(HttpServletRequest req, HttpServletResponse resp) {
        this(req, resp, (String) null);
    }   

    
    public ServletSseSubscriber(HttpServletRequest req, HttpServletResponse resp, String initalComment) {
        this(req, resp, Duration.ofSeconds(25), initalComment);
    }   

    
    public ServletSseSubscriber(HttpServletRequest req, HttpServletResponse resp, Duration keepAlivePeriod) {
        this(req, resp, keepAlivePeriod, null);
    }
    
    public ServletSseSubscriber(HttpServletRequest req, HttpServletResponse resp, Duration keepAlivePeriod, String initalComment) {
        if (!req.isAsyncStarted()) {
            req.startAsync().setTimeout(Long.MAX_VALUE);  // tomcat 7 default is 30 sec 
        }
        
        try {
            this.channel = new SseOutboundChannel(resp.getOutputStream(), error -> onError(error), keepAlivePeriod, initalComment);
        } catch (IOException ioe) {
            throw new RuntimeException(ioe);
        }
    }   


    @Override
    public void onSubscribe(Subscription subscription) {
        subscriptionRef.set(subscription);
        subscriptionRef.get().request(25);  // request next 25 
    }

    
    
    @Override
    public void onNext(ServerSentEvent event) {
        channel.writeEventAsync(event)           
               .thenAccept(written -> subscriptionRef.get().request(1)); 
    }

    @Override
    public void onError(Throwable t) {
        LOG.debug("error on source stream. stop streaming " + t.toString());
        close();
    }
  
    @Override
    public void onComplete() {
        close();
    }
    

    private void close() {
        if (isOpen.getAndSet(false)) {
            subscriptionRef.get().cancel();
            channel.close();          
        }
    }
} public class Pair<F,S> {

    private final F first;
    private final S second;


    private Pair(F first, S second) {
        this.first = first;
        this.second = second;
    }

    /**
     * @return the first item in the pair
     */
    public F getFirst() {
        return first;
    }
    
    /**
     * @return the second item in the pair
     */
    public S getSecond() {
        return second;
    }

    /**
     * @param first the first item to store in the pair
     * @param second the second item to store in the pair
     * @param <S> the type of the first item
     * @param <T> the type of the second item
     * @return a new pair wrapping the two items
     */
    public static <F, S> Pair<F, S> of(F first, S second) {
        return new Pair<>(first,second);
    }
    
    
    @Override
    public String toString() {
        return "[" + first + ", " + second + "]";
    }
    
    @Override
    public int hashCode() {
        return Objects.hashCode(first, second);
    }
    
    @SuppressWarnings("rawtypes")
    @Override
    public boolean equals(Object other) {

        if ((other != null) && (other instanceof Pair)) {
            Pair otherPair = (Pair) other;
            return Optional.ofNullable(otherPair.getFirst()).equals(Optional.ofNullable(this.getFirst())) &&
                   Optional.ofNullable(otherPair.getSecond()).equals(Optional.ofNullable(this.getSecond()));
            
        } else {
            return false;
        }
    }
}
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.Locale;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.Predicate;
import javax.json.Json;
import javax.json.JsonArrayBuilder;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.BadRequestException;
import javax.ws.rs.DefaultValue;
import javax.ws.rs.GET;
import javax.ws.rs.HeaderParam;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriInfo;
import org.apache.avro.generic.GenericData.Record;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import com.google.common.base.Charsets;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import com.unitedinternet.mam.incubator.hammer.problem.StdProblem;
import net.oneandone.incubator.neo.collect.Immutables;
import net.oneandone.incubator.neo.freemarker.Page;
import net.oneandone.incubator.neo.hypermedia.LinksBuilder;
import net.oneandone.reactive.ReactiveSource;
import net.oneandone.reactive.kafka.CompletableKafkaProducer;
import net.oneandone.reactive.kafka.KafkaMessage;
import net.oneandone.reactive.kafka.KafkaMessageId;
import net.oneandone.reactive.kafka.KafkaMessageIdList;
import net.oneandone.reactive.kafka.KafkaSource;
import net.oneandone.reactive.kafka.avro.json.AvroMessage;
import net.oneandone.reactive.kafka.avro.json.AvroMessageMapperRepository;
import net.oneandone.reactive.rest.container.ResultConsumer;
import net.oneandone.reactive.sse.ServerSentEvent;
import net.oneandone.reactive.sse.servlet.ServletSseSubscriber;
import net.oneandone.reactive.utils.Pair;
import rx.Observable;
import rx.RxReactiveStreams;
/*
 * Copyright 1&1 Internet AG, https://github.com/1and1/
 * 
 * Licensed 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.
 */
package net.oneandone.reactive.kafka.rest;












@Path("/")
public class BusinesEventResource {
    
    private final AvroMessageMapperRepository mapperRepository; 
    private final CompletableKafkaProducer<String, byte[]> kafkaProducer;
    private final KafkaSource<String, byte[]> kafkaSource;

    

    @Autowired
    public BusinesEventResource(final CompletableKafkaProducer<String, byte[]> kafkaProducer,
                                final KafkaSource<String, byte[]> kafkaSource,
                                final AvroMessageMapperRepository avroMessageMapperRepository) {
        
        this.kafkaProducer = kafkaProducer;
        this.kafkaSource = kafkaSource;
        this.mapperRepository = avroMessageMapperRepository;
    }
    
    

    @GET
    @Path("/")
    @Produces("application/json; qs=0.8")
    public ImmutableMap<String, ImmutableMap<String, Object>> getRootJson(final @Context UriInfo uriInfo) {

        return ImmutableMap.of("_links", LinksBuilder.create(uriInfo).withHref("topics").build());
    }
 
 
    @GET
    @Path("/")
    @Produces("text/html; qs=0.4")
    public Page getRootHtml(final @Context UriInfo uriInfo, 
                            final @Context HttpServletRequest httpRequest) {
        
        return new Page("/net/oneandone/reactive/kafka/rest/root.ftl")
                      .withModelData("self", uriInfo.getBaseUri())
                      .withModelData("base", uriInfo.getBaseUri().toString().substring(0, uriInfo.getBaseUri().toString().length() - httpRequest.getServletPath().length()));
    }
    
    
  
    
    @GET
    @Path("/topics")
    @Produces("application/vnd.ui.mam.eventservice.topic.list+json; qs=0.8")
    public TopicsRepresentation getTopicsJson(final @Context UriInfo uriInfo, 
                                              final @QueryParam("q.topic.name.eq") String topicname) {

        Predicate<String> filter = (topicname == null) ? (name -> true) : (name -> name.equals(topicname));
        
        return new TopicsRepresentation(uriInfo, "topics", ImmutableSet.copyOf(kafkaSource.listTopics()
                                                                                          .stream()
                                                                                          .filter(filter)
                                                                                          .map(name -> getTopicJson(uriInfo, name))
                                                                                          .collect(Immutables.toList()))); 
    }
    
    
    @GET
    @Path("/topics")
    @Produces("text/html; qs=0.4")
    public Page getTopicsHtml(final @Context UriInfo uriInfo) {
        
        return new Page("/net/oneandone/reactive/kafka/rest/topiclist.ftl")
                      .withModelData("self", uriInfo.getBaseUri())
                      .withModelData("topicnames", kafkaSource.listTopics());
    }
    
    
    @GET
    @Path("/topics/{topicname}")
    @Produces("application/vnd.ui.mam.eventservice.topic+json; qs=0.8")
    public TopicRepresentation getTopicJson(final @Context UriInfo uriInfo,
                                            final @PathParam("topicname") String topicname) {
        
        return new TopicRepresentation(uriInfo, "topics", topicname);
    }
        
    
    @GET
    @Path("/topics/{topicname}")
    @Produces("text/html; qs=0.4")
    public Page getTopicHtml(final @Context UriInfo uriInfo, 
                             final @PathParam("topicname") String topicname) {
        
        return new Page("/net/oneandone/reactive/kafka/rest/topic.ftl")
                      .withModelData("base", uriInfo.getBaseUri())
                      .withModelData("topicname", topicname);
    }
        
    
    
    

    @GET
    @Path("/topics/{topic}/schemas")
    @Produces("text/plain; qs=0.8")
    public String getRegisteredSchemasPlain() {
        
        return Joiner.on("\r\n").join(mapperRepository.getRegisteredSchemasAsText()
                                                      .entrySet()
                                                      .stream()
                                                      .map(entry -> ("== " + entry.getKey() +  
                                                                     " ==\r\n" + entry.getValue() +
                                                                     "\r\n\r\n\r\n"))
                                                      .collect(Immutables.toList()));
    }
    
  
    @GET
    @Path("/topics/{topic}/schemas")
    @Produces("text/html; qs=0.4")
    public Page getRegisteredSchemasHtml() throws IOException {

        ImmutableMap<String, String> schemas = mapperRepository.getRegisteredSchemas()
                                                               .entrySet()
                                                               .stream()
                                                               .collect(Immutables.toMap(e -> e.getKey(), 
                                                                                         e -> (e.getValue().getDoc() == null) ? "" 
                                                                                                                              : e.getValue().getDoc()));
        ImmutableSet<String> mimeTypes = ImmutableSet.copyOf(Sets.newTreeSet(schemas.keySet()));
        
        return new Page("/net/oneandone/reactive/kafka/rest/schemalist.ftl")
                    .withModelData("mimeTypes", mimeTypes)
                    .withModelData("schemas", schemas);
    }
    
    

    @GET
    @Path("/topics/{topic}/erroneousschemas")
    @Produces("text/html; qs=0.4")
    public Page getErroneousSchemasHtml() throws IOException {
                                
        return new Page("/net/oneandone/reactive/kafka/rest/erroneousschemalist.ftl")
                    .withModelData("erroneousSchemas", mapperRepository.getErroneousSchemas());
    }
    
    
    

    @GET
    @Path("/topics/{topic}/schemas/{type}/{subtype}")   
    @Produces("text/plain")
    public Response getRegisteredSchemaPlain(final @PathParam("type") String type, 
                                             final @PathParam("subtype") String subtype) {
        
        String schema = mapperRepository.getRegisteredSchemasAsText()
                                        .get(type + "/" + subtype);
        
        return  (schema == null) ? StdProblem.newNotExists().toResponse()