사용자 제어 서비스

// ChannelAcceptHandler 샘플 코드
public class UCSChannelAcceptHandler extends ChannelAcceptHandler {
    private ProObjectLogger     logger = ChannelEventLogger.getLogger();
    private ServerSocketChannel serverSocketChannel;
    private UCSWorkerThread     workerThread;

    public UCSChannelAcceptHandler(UCSWorkerThread workerThread) {
        this.workerThread = workerThread;
    }

    @Override
    public void init() throws IOException {
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
    }

    @Override
    public void accept(int port, Object additionalInfo) {
        try {
            String acceptAddress = getAcceptAddress();
            if(acceptAddress == null) {
                serverSocketChannel.bind(new InetSocketAddress(port));
            } else {
                serverSocketChannel.bind(new InetSocketAddress(acceptAddress, port));
            }
            workerThread.registerChannel(serverSocketChannel, this, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            close();
            throw new RuntimeException(e);
        }
    }

    @Override
    public void onAccept(AbstractSelectableChannel serverChannel, AbstractSelectableChannel clientChannel) {
        try {
            SocketChannel socketChannel = (SocketChannel) clientChannel;
            socketChannel.configureBlocking(false);

            UCSChannelEventHandler channelEventHandler = new UCSChannelEventHandler(socketChannel, workerThread);
            channelEventHandler.init(SelectionKey.OP_READ);
            workerThread.setChannelEventHandler(channelEventHandler);

            logger.fine("[UCSChannelAcceptHandler] onAccept success");
        } catch (Throwable e) {
            logger.severe("[UCSChannelAcceptHandler] Establishing connection is failed...", e);
        }
    }

    @Override
    public void close() {
        if(logger.isLoggable(Level.INFO)) {
            logger.info("[UCSChannelAcceptHandler] close");
        }

        if(serverSocketChannel != null && serverSocketChannel.isRegistered()) {
            try {
                serverSocketChannel.close();
                workerThread.unregisterChannel(serverSocketChannel);
                serverSocketChannel = null;
            } catch (IOException e) {
                logger.severe("[UCSChannelAcceptHandler][{0}] Channel can not be closed!", this.getClass().getSimpleName(), e);
            }

            if(logger.isLoggable(Level.INFO)) {
                logger.info("[UCSChannelAcceptHandler] Server Channel is closed.");
            }
        }
    }
}

// ChannelEventHandler 샘플 코드
public class UCSChannelEventHandler extends ChannelEventHandler {
    private     UCSWorkerThread             workerThread;
    private     ByteBuffer                  readBuffer;
    private     ByteBuffer                  writeBuffer;
    private     Map<SourceChannel, String>  timeoutChannelMap;
    private     Pipe                        pollingChannel;
    private     Pipe                        waitingChannel;

    public UCSChannelEventHandler(SocketChannel socketChannel, UCSWorkerThread workerThread) throws IOException {
        setChannel(socketChannel);
        this.workerThread = workerThread;
        readBuffer = ByteBuffer.allocate(1024);
        writeBuffer = ByteBuffer.allocate(1024);
        timeoutChannelMap = new HashMap<>();
    }

    @Override
    public void init(int interestOps) throws IOException {
        setInterestOps(interestOps);
        try {
            SocketChannel socket = (SocketChannel)getChannel();
            if(socket.isRegistered() == false) {
                workerThread.registerChannel(socket, this, interestOps);

                // * 주기적인 health check를 위한 처리
                pollingChannel = Pipe.open();
                pollingChannel.source().configureBlocking(false);
                workerThread.registerChannel(pollingChannel.source(), this, interestOps);
                workerThread.registerTimeout(pollingChannel.source(), 10*1000L);
                timeoutChannelMap.put(pollingChannel.source(), "POLLING");

                // * write timeout을 위한 처리
                waitingChannel = Pipe.open();
                waitingChannel.source().configureBlocking(false);
                workerThread.registerChannel(waitingChannel.source(), this, interestOps);
                timeoutChannelMap.put(waitingChannel.source(), "WAITING");

                getLogger().info("[UCSChannelEventHandler] init() success");
            }
        } catch(Throwable e) {
            getLogger().severe("[UCSChannelEventHandler] is failed to initialize!", e);
            if(getChannel() != null) {
                try {
                    getChannel().close();
                } catch(Throwable ignore) {}
            }
        }
    }

    @Override
    public void onRead(AbstractSelectableChannel channel) throws IOException {
        SocketChannel socketChannel = (SocketChannel)channel;

        try {
            if(readBuffer.hasRemaining()) {
                if(socketChannel.read(readBuffer) < 0) {
                    close();
                    return;
                }
            }
            readBuffer.flip();
            String data = new String(readBuffer.array(), readBuffer.position(), readBuffer.limit());

            // * UCS 수신
            if(getLogger().isLoggable(Level.INFO)) {
                getLogger().info("[UCSChannelEventHandler] onRead() : {0}", data);
            }

            /**
             * Client로부터 read했을 때의
             * user logic 작성
             */

            if(data.equalsIgnoreCase("efgh")) {
                // * 클라이언트 -> 서비스 -> FEP 시나리오 (SVC -> UCS -> FEP -> UCS -> SVC)
                ServiceManager.ucsCall(data);
            } else if (data.equalsIgnoreCase("ijkl")) {
                // * FEP -> 서비스 시나리오 (FEP -> UCS -> SVC -> UCS -> FEP)
                StringDataObject input = new StringDataObject();
                input.setValue("sync/receive");
                input.setDesc("mnop");

                RequestContext requestContext = ServiceManager.getServiceRequestContext("test.main.UCSCommonInterfaceCall");
                Header header = new Header();
                header.set("TEST", "test");
                requestContext.getRequest().setHeader(header);

                ServiceManager.ucsCall(requestContext, "test.main.UCSCommonInterfaceCall", input);
            } else if (data.equalsIgnoreCase("uvwx")) {
                // * 랑데뷰 시나리오
                // * SVC1 -> UCS -> FEP1
                // * FEP2 -> UCS -> SVC2 -> SVC1 랑데뷰 -> UCS -> FEP
                StringDataObject input = new StringDataObject();
                input.setValue("matchingKey");
                ServiceManager.ucsCall("test.main.UCSRendezvousSub", input);
            }  else if (data.equalsIgnoreCase("sticky")) {
                // * 전문 역전 방지 (Sticky Service) 검증 시나리오
                StringDataObject input = new StringDataObject();
                input.setValue(data);
                ServiceManager.ucsCall("test.main.UCSCommonInterfaceCall", input);
            }

            readBuffer.clear();
        } catch(Exception e) {
            close();
            if(getLogger().isLoggable(Level.SEVERE)) {
                getLogger().severe("[UCSChannelEventHandler] onRead() fail..", e);
            }
        }
    }

    @Override
    public void onWrite(AbstractSelectableChannel channel, Object data, UserContext userContext, long requestId) throws IOException {
        String logMessage = null;
        if(data instanceof String) {
            // 실제로 보낼 데이터 조립
            writeBuffer.put(((String) data).getBytes());
            logMessage = data.toString();
        } else if (data instanceof DataObject) {
            DataObject dobj = (DataObject)data;
            String value = dobj.get("value");
            writeBuffer.put(value.getBytes());
            logMessage = value.toString();
        }

        if(getLogger().isLoggable(Level.INFO)) {
            getLogger().info("[UCSChannelEventHandler] onWrite() : {0}", logMessage);
        }

        workerThread.interestOpsChannel(channel, SelectionKey.OP_WRITE);
    }

    @Override
    public void onWritable(AbstractSelectableChannel channel) {
        SocketChannel socketChannel = (SocketChannel)channel;

        try {
            writeBuffer.flip();
            if(socketChannel.write(writeBuffer) < 0) {
                close();
            }
            writeBuffer.clear();
            workerThread.interestOpsChannel(socketChannel, interestOps);
            workerThread.registerTimeout(waitingChannel.source(), 10*1000L);
        } catch(Exception e) {
            close();
        }
    }

    @Override
    public void onTimeout(AbstractSelectableChannel channel) throws Exception {
        String value = timeoutChannelMap.get(channel);

        if(getLogger().isLoggable(Level.INFO)) {
            getLogger().info("[UCSChannelEventHandler] onTimeout() : {0}", value);
        }

        switch (value) {
        case "POLLING":
            if(pollingChannel != null) {
                // * FEP로 POLL 전문을 write 하면 됨
                workerThread.registerTimeout(pollingChannel.source(), 10*1000L);
            }
            break;
        case "WAITING":
            if(waitingChannel != null) {
                // * write했을 때 timeout이 발생한 경우에 대한 처리 로직
            }
            break;
        default:
            break;
        }
    }

    @Override
    public void onClose(AbstractSelectableChannel channel) throws Exception {
        if(getLogger().isLoggable(Level.INFO)) {
            getLogger().info("[UCSChannelEventHandler] onClose()");
        }

        if(channel != null && channel.isRegistered()) {
            workerThread.unregisterChannel(channel);
            channel.close();
            setChannel(null);

            workerThread.unregisterChannel(pollingChannel.source());
            try {
                pollingChannel.source().close();
                pollingChannel.sink().close();
            } catch(Throwable ignore) {}

            pollingChannel = null;

            workerThread.unregisterChannel(waitingChannel.source());
            try {
                waitingChannel.source().close();
                waitingChannel.sink().close();
            } catch(Throwable ignore) {}
            waitingChannel = null;

            if(getLogger().isLoggable(Level.INFO)) {
                getLogger().info("[UCSChannelEventHandler] Channel is closed.");
            }
        }
    }

    @Override
    public void onError(AbstractSelectableChannel channel, Throwable error) {
        if(getLogger().isLoggable(Level.SEVERE)) {
            getLogger().severe("[UCSChannelEventHandler] onError()", error);
        }
    }
}

public class UCSMainThread extends UCSWorkerThread {
    public UCSMainThread(ServiceWorkerThreadPool threadPool, ServiceGroupManager serviceManager, ThreadGroup threadGroup, int threadId, int priority) throws Throwable {
        super(threadPool, serviceManager, threadGroup, threadId, priority);

        UCSChannelAcceptHandler acceptHandler = new UCSChannelAcceptHandler(this);
        setChannelAcceptHandler(acceptHandler);
        acceptHandler.init();

        int serverPort = serviceManager.getUCSInfoManager().getPort(threadPool.getName());
        ChannelEventLogger.getLogger().info("===== UCSMainThread -> port : {0}  =====", serverPort);
        acceptHandler.accept(serverPort, null);

        ContextDataObject context = serviceManager.getUCSInfoManager().getDetail(threadPool.getName());
        ChannelEventLogger.getLogger().info("===== UCSMainThread -> detail : {0}  =====", context);
    }
}

//샘플 ServiceObject
public class UCSCommonInterfaceCall implements ServiceObject<StringDataObject, StringDataObject>{

    @Override
    public StringDataObject service(StringDataObject input) throws Throwable {
        String value = input.getValue();
        StringDataObject output = null;

        switch(value) {
        case "sync/send" :
            //* rcall과 비슷하게 동작한다고 생각하면 됨
            ServiceManager.ucsCall("abcd", "response", "response");
            break;
        case "sync/receive" :
            Header header = (Header) ServiceManager.getCurrentRequestContext().getRequest().getHeader();
            output = new StringDataObject();
            output.setValue("header > " + header.get("TEST") + " > " + input.getDesc());
            break;
        case "sticky" :
            int count = 3;
            WaitObject[] waitObjects = new WaitObject[count];
            for(int i=0;i < count;i++) {
                StringDataObject stickyInput = new StringDataObject();
                stickyInput.setValue(String.valueOf(i));
                stickyInput.setDesc("3");
                waitObjects[i] = ServiceManager.acall("test.main.StickyServiceCall", stickyInput, "ucs");
            }

            WaitObject and = WaitObject.And(waitObjects);
            try {
                WaitObject[] result = (WaitObject[])and.get();
                String outputMessage = "callSequence;threadId;time";
                for(int i = 0; i < result.length; i++) {
                    StringDataObject stickyOutput = (StringDataObject)result[i].get();
                    outputMessage = outputMessage + " > " + stickyOutput.getValue();
                }

                output = new StringDataObject();
                output.setValue(outputMessage);
            } catch(Throwable e) {
                if(logger.isLoggable(Level.WARNING)) {
                    logger.warning("[UCSCommonInterfaceCall] StickyServiceCall gave me a exception!", e);
                }
                throw e;
            }
        default :
            break;
        }

        return output;
    }

    @RCallReturn
    public Object response(Object result) {
        StringDataObject dtoOut = new StringDataObject();
        dtoOut.setValue(result.toString());
        return dtoOut;
    }
}

//샘플 ServiceExecutor
public class UCSCommonInterfaceCallExecutor extends ServiceExecutor {

    public UCSCommonInterfaceCallExecutor() {
        serviceObject = new UCSCommonInterfaceCall();
    }

    @SuppressWarnings("unchecked")
    @Override
    public Object execute(Object input, String methodName) throws Throwable {
        switch(methodName) {
        case "response" :
            return ((UCSCommonInterfaceCall)serviceObject).response(input);
        default :
            return serviceObject.service(input);
        }
    }
}

//샘플 ServiceObject
public class StickyServiceCall implements ServiceObject<StringDataObject, StringDataObject> {
    @Override
    public StringDataObject service(StringDataObject input) throws Throwable {
        String time            = String.valueOf(System.currentTimeMillis());
        String callSequence    = input.getValue();
        String threadId        = ((ServiceWorkerThread)Thread.currentThread()).getThreadId();
        int    sleep           = StringUtil.isEmpty(input.getDesc()) ? 1 : Integer.parseInt(input.getDesc());

        Thread.sleep(sleep * 1000);

        StringBuilder builder = new StringBuilder();
        builder.append(callSequence).append(";").append(threadId).append(";").append(time);

        StringDataObject output = new StringDataObject();
        output.setValue(builder.toString());
        output.setDesc("callSequence;threadId;time");

        if(ServiceLogger.getLogger().isLoggable(Level.INFO)) {
            ServiceLogger.getLogger().info("[StickyServiceCall] output : {0}", output);
        }

        return output;
    }
}

//샘플 ServiceObject
public class UCSRendezvousMain implements ServiceObject<StringDataObject, StringDataObject> {
    private String  result;

    @Override
    public StringDataObject service(StringDataObject input) throws Throwable {
        result = input.getValue() + " > SM.RENDEZVOUS";

        WaitObject wo = ServiceManager.rendezvous(input.getValue(), "test.main.UCSRendezvousMain" , "test.main.UCSRendezvousSub");
        ServiceManager.ucsCall(wo, "qrst", "result", "result");
        return null;
    }

    @RCallReturn
    public Object result(WaitObject waitObject) throws Throwable {
        StringDataObject callOutput = (StringDataObject)((WaitObject[])waitObject.get())[0].get();
        result += " > " + callOutput.getValue() + " > UCS_MAIN";

        StringDataObject output = new StringDataObject();
        output.setValue(result);
        return output;
    }
}

//샘플 ServiceExecutor
public class UCSRendezvousMainExecutor extends ServiceExecutor {
    public UCSRendezvousMainExecutor() {
        serviceObject = new UCSRendezvousMain();
    }

    @Override
    public Object execute(Object input, String methodName) throws Throwable {
        switch (methodName) {
        case BaseServiceExecutor.DEFAULT_SERVICE_METHOD:
            return ((UCSRendezvousMain)serviceObject).service((StringDataObject)input);
        case "result":
            return ((UCSRendezvousMain)serviceObject).result((WaitObject)input);
        }

        throw new Exception();
    }
}

//샘플 ServiceObject
public class UCSRendezvousSub implements ServiceObject<StringDataObject, StringDataObject> {
    @Override
    public StringDataObject service(StringDataObject input) throws Throwable {
        String value = input.getValue();
        ServiceManager.rendezvous(input.getValue(), "test.main.UCSRendezvousMain" , "test.main.UCSRendezvousSub");

        StringDataObject output = new StringDataObject();
        output.setValue(value + " > UCS_SUB");
        return output;
    }
}

//샘플 ServiceExecutor
public class UCSRendezvousSubExecutor extends ServiceExecutor {
    public UCSRendezvousSubExecutor() {
        serviceObject = new UCSRendezvousSub();
    }

    @SuppressWarnings("unchecked")
    @Override
    public Object execute(Object input, String methodName) throws Throwable {
        return serviceObject.service(input);
    }
}

HTTP Method : POST
URL    : http://IP:PORT/proobject/ucs/UCSInfo
INPUT  : { "dto" :
              { "serverName" : "pomain",
                "poolId" : "abcd",
                "port" : 9997,
                "portDesc" : "sample server port",
                "detail" : "{\"key\" : \"value\"}",
                "status" : "RUNNING"
              }
           }
OUTPUT  : { "dto" :
              { "count" : 1,
                "output" : "Insert Success"
               }
            }

 HTTP Method : GET
URL    : http://IP:PORT/proobject/ucs/serverName/pomain/application/test/serviceGroup/main/UCSInfo?poolId=abcd&port=9997&status=RUNNING
OUTPUT : { "dto" :
              { "info" :
                 [{"poolId" : "abcd",
                   "serverName" : "pomain",
                   "application" : "test",
                   "serviceGroup" : "main",
                   "poolName" : "ucsMainPool",
                   "threadClassName" : "...",
                   "threadMinCount" : 1,
                   "threadCoreCount" : 1,
                   "threadMaxCount" : 1,
                   "status" : "RUNNING",
                   "port" : 9997,
                   "portDesc" : "sample server port",
                   "detail" : "{ \"key\" : \"value\" }"
                }]
             }
           }

HTTP Method : PUT
URL     : http://IP:PORT/proobject/ucs/UCSInfo
INPUT   : { "dto" :
               { "serverName" : "pomain",
                 "application" : "test",
                 "serviceGroup" : "main",
                 "poolName" : "ucsMainPool",
                 "port" : 9998,
                 "portDesc" : "update sample server port",
                 "detail" : "{\"updateKey\" : \"updateValue\"}"
               }
           }
OUTPUT  : { "dto" :
              { "count" : 1,
                "output" : "Update Success"
              }
           }

HTTP Method : Delete
URL         : http://IP:PORT/proobject/ucs/UCSInfo
INPUT       : { "dto" : { "serverName" : "pomain",
                          "application" : "test",
                          "serviceGroup" : "main",
                          "poolName" : "ucsMainPool" } }
OUTPUT      : { "dto" : { "count" : 1,
                          "output" : "Delete Success" } }