/*
 * ⁣​
 * YADS
 * ⁣⁣
 * Copyright (C) 2015 - 2016 srgg
 * ⁣⁣
 * 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 com.github.srgg.yads.impl;

import com.github.srgg.yads.api.IStorage;
import com.github.srgg.yads.impl.api.context.Subscribe;
import com.github.srgg.yads.api.message.Messages;
import com.github.srgg.yads.impl.api.context.OperationExecutionContext;
import com.github.srgg.yads.impl.api.context.StorageNodeContext;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import com.github.srgg.yads.api.messages.StorageOperation;

import static com.google.common.base.Preconditions.checkState;

/**
 *  @author Sergey Galkin <srggal at gmail dot com>
 */
public class StorageNode extends AbstractNode<StorageNodeContext> {
    // TODO: consider to make it autogenerated
    public enum StorageState { NEW, STARTED, RUNNING, RECOVERING, STOPPED, FAILED }

    private final IStorage storage;

    private AbstractProcessingCycle<StorageOperation> processingCycle =
            new AbstractProcessingCycle<StorageOperation>() {

        @Override
        protected OperationExecutionContext<StorageOperation> getContextFor(final StorageOperation op) {
            final StorageNodeContext ctx = context();
            checkState(ctx != null, "Storage context can't be null");
            return ctx.contextFor(op);
        }
    };

    public StorageNode(final String nid, final IStorage s) {
        super(nid, Messages.NodeType.Storage);
        this.storage = s;
    }

    @Override
    protected void onStateChanged(final String old, final String current) {
        super.onStateChanged(old, current);
        if (State.RUNNING.name().equals(current)) {
            processingCycle.start();
        } else if (State.STOPPED.name().equals(old)) {
            processingCycle.stop();
        }
    }

    @Override
    protected Map<String, Set<String>> createTransitions() {
        return TransitionMatrixBuilder.create(super.createTransitions())
                .add(StorageState.STARTED, StorageState.RECOVERING)
                .add(StorageState.FAILED, StorageState.RECOVERING)
                .add(StorageState.RECOVERING, StorageState.RUNNING, StorageState.FAILED, StorageState.STOPPED)
            .build();
    }

    @Subscribe
    public void onStorageRequest(final StorageOperation operation) throws Exception {
        final String s = getState();
        if (!StorageState.RUNNING.name().equals(s) && !StorageState.RECOVERING.name().equals(s)) {
            throw new IllegalStateException(
                    String.format("Can't execute request, wrong getState '%s'", s)
            );
        }
        processingCycle.enqueue(operation);
    }


//    protected void onRoleChanged(EnumSet<ControlMessage.Role> old, EnumSet<ControlMessage.Role> current) {
//        logger().info("[ROLE] was changed from '{}' to '{}'", old, current);
//    }
//
//    @Subscribe
//    public boolean onCoordination(ControlMessage m) {
//        logger().info("[CTRL:{}]  '{}'...", m.getSender(), m.getId().toString());
//        if (m.getType().contains(ControlMessage.Type.SetRole)) {
//            checkState(m.getRoles() != null && !m.getRoles().isEmpty(), "Roles can't be null");
//            final EnumSet<ControlMessage.Role> old = roles;
//
//            if (!m.getRoles().equals(old)) {
//                roles = m.getRoles();
//                onRoleChanged(old, m.getRoles());
//            }
//        }
//
//        if (m.getType().contains(ControlMessage.Type.SetChain)) {
//            // TODO
//        }

//        if (m.getType().contains(ControlMessage.Type.SetState)) {
//            setState(m.getState());
//        }
//
//        return true;
//    }

    private abstract class AbstractProcessingCycle<T extends StorageOperation> implements Runnable {
        private Thread thread;
        private LinkedBlockingQueue<T> queuedRequests = new LinkedBlockingQueue<>();

        public void start() {
            logger().debug("is about to start processing cycle");
            if (thread != null) {
                throw new IllegalStateException(
                        String.format("Node '%s': processing cycle is already started", getId())
                );
            }

            thread = new Thread(this);
            thread.start();
            logger().info("processing cycle has been STARTED");
        }

        public void stop() {
            if (thread == null) {
                logger().warn("Processing cycle can't be stopped since it hasn't been started");
            } else {
                logger().debug("processing cycle is about to stop");
                thread.interrupt();
                try {
                    thread.join();
                } catch (InterruptedException ex) {
                }
                logger().info("processing cycle has been STOPPED");
            }
        }

        public void enqueue(final T operation) {
            queuedRequests.offer(operation);
            logger().debug("storage operation enqueued {}:'{}'",
                    operation.getType(), operation.getId());
        }

        protected abstract  OperationExecutionContext<T> getContextFor(T op);

        @Override
        public void run() {
            Thread.currentThread().setName("node-" + getId() + "-prcsngcycle");
            for (;;) {
                try {
                    final T op = queuedRequests.peek();
                    if (op == null) {
                        // TODO: re-introduce as a config parameter
                        Thread.sleep(100L);
                        continue;
                    }

                    if (!State.RUNNING.name().equals(getState())) {
                        logger().debug("is not to be running, processing cycle will be terminated");
                        break;
                    }

                    final OperationExecutionContext<T> ctx = getContextFor(queuedRequests.poll());
                    assert ctx != null;
                    T operation = null;
                    try {
                        operation = ctx.operation();
                        final Object result = storage.process(operation);

                        logger().debug("storage operation succeeded {}:'{}'",
                                getId(), operation.getType(), operation.getId());

                        ctx.ackOperation(result);
                    } catch (Exception ex) {
                        logger().error(
                                String.format("storage operation %s:'%s' is failed to process",
                                        operation == null ? null : operation.getType(),
                                        operation == null ? null : operation.getId()
                                    ),
                                ex
                            );

                        ctx.failureOperation(ex);
                    }
                } catch (InterruptedException ex) {
                    logger().debug("got InterruptedException, processing cycle will be terminated");
                    break;
                }
            }

            logger().info("processing cycle has been terminated");
        }
    }
}

@@ -2183,24 +2183,31 @@
         if (
+Storage
 State.RUNNIN
@@ -2290,16 +2290,23 @@
 lse if (
+Storage
 State.ST
@@ -2321,27 +2321,31 @@
 me().equals(
-old
+current
 )) %7B%0A       
@@ -5324,32 +5324,139 @@
 Exception ex) %7B%0A
+                    logger().debug(%22Got InterruptedException during JOIN to the processing cycle thread%22);%0A
                 

Use StorageState in StorageNode