[RM] Feature Create abstract classes for mina server.
ADD
BackEnd/Core/sailfish-core/src/main/java/com/exactpro/sf/services/codecs/ICodecSettings.java
+/*******************************************************************************
+ * Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+ *
+ * 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.exactpro.sf.services.codecs;
+
+import com.exactpro.sf.common.util.ICommonSettings;
+import com.exactpro.sf.configuration.suri.SailfishURI;
+
+/**
+ * Interface for base codec settings
+ */
+public interface ICodecSettings extends ICommonSettings {
+
+ SailfishURI getDictionaryURI();
+ String getFilterValues();
+
+}

ADD
BackEnd/Core/sailfish-core/src/main/java/com/exactpro/sf/services/mina/AbstractMINATCPServer.java
+/*******************************************************************************
+ * Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+ *
+ * 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.exactpro.sf.services.mina;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.mina.core.service.IoHandler;
+import org.apache.mina.core.service.IoHandlerAdapter;
+import org.apache.mina.core.session.IoSession;
+import org.apache.mina.filter.codec.ProtocolCodecFilter;
+
+import com.exactpro.sf.common.codecs.AbstractCodec;
+import com.exactpro.sf.common.codecs.CodecFactory;
+import com.exactpro.sf.common.messages.IMessage;
+import com.exactpro.sf.services.AbstractInitiatorService;
+import com.exactpro.sf.services.IAcceptorService;
+import com.exactpro.sf.services.ISession;
+import com.exactpro.sf.services.ServiceException;
+import com.exactpro.sf.services.ServiceHandlerException;
+import com.exactpro.sf.services.WrapperNioSocketAcceptor;
+
+public abstract class AbstractMINATCPServer extends AbstractInitiatorService implements IoHandler, IAcceptorService {
+
+ protected final AtomicReference<MINAServerSession> serverSession;
+ protected final AtomicReference<WrapperNioSocketAcceptor> acceptor;
+ protected final Map<IoSession, MINASession> sessions;
+
+ public AbstractMINATCPServer() {
+ sessions = new ConcurrentHashMap<>();
+ serverSession = new AtomicReference<>();
+ acceptor = new AtomicReference<>();
+ }
+
+ @Override
+ public List<ISession> getSessions() {
+ return new ArrayList<>(sessions.values());
+ }
+
+ @Override
+ protected void internalStart() throws IOException {
+ WrapperNioSocketAcceptor tmpAcceptor = new WrapperNioSocketAcceptor(taskExecutor);
+
+ try {
+ configureAcceptor(tmpAcceptor);
+ } catch (Exception e) {
+ throw new ServiceException("Can`t configure server`s acceptor", e);
+ }
+
+ CodecFactory codecFactory = new CodecFactory(serviceContext, messageFactory, dictionary, getCodecClass(), getSettings());
+ tmpAcceptor.setHandler(this);
+ tmpAcceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(codecFactory));
+ tmpAcceptor.bind(new InetSocketAddress(getSettings().getHost(), getSettings().getPort()));
+
+ acceptor.set(tmpAcceptor);
+ serverSession.set(createServerSession());
+ }
+
+ @Override
+ public AbstractMINATCPServerSettings getSettings() {
+ return (AbstractMINATCPServerSettings) settings;
+ }
+
+ @Override
+ public MINAServerSession getSession() {
+ return serverSession.get();
+ }
+
+ @Override
+ public void connect(){
+ //TODO: Must throws exception and use 'accept' or other method for server.
+ //throw new UnsupportedOperationException("Server can't connect to anyone but it can accept incoming connection");
+ }
+
+ @Override
+ protected void disposeResources() {
+
+ MINAServerSession session = serverSession.getAndSet(null);
+
+ try {
+ if (session != null && session.isConnected()) {
+ session.close();
+ }
+ } catch (RuntimeException e) {
+ logger.warn("Server session didn`t close", e);
+ }
+
+ WrapperNioSocketAcceptor localAcceptor = acceptor.getAndSet(null);
+ try {
+ if (localAcceptor != null) {
+ localAcceptor.unbind();
+ localAcceptor.dispose();
+ }
+ } catch (RuntimeException e) {
+ logger.warn("Acceptor didn`t unbind or/and dispose", e);
+ }
+
+ super.disposeResources();
+ }
+
+ @Override
+ protected String getEndpointName() {
+ return String.format("Server for %s", serviceName);
+ }
+
+ @Override
+ public void sessionCreated(IoSession session) {
+ logger.debug("Session created. Session: " + session);
+ }
+
+ @Override
+ public void sessionOpened(IoSession session) throws ServiceHandlerException {
+ MINASession minaSession = createMINASession(session);
+ onSessionOpen(minaSession);
+ sessions.put(session, minaSession);
+ handler.sessionOpened(serverSession.get());
+
+ logger.debug("sessionOpened - {} session is opened", session);
+ }
+
+ @Override
+ public void sessionClosed(IoSession session) throws Exception {
+ onSessionClosed(sessions.remove(session));
+ handler.sessionClosed(serverSession.get());
+ }
+
+ @Override
+ public void exceptionCaught(IoSession session, Throwable cause) {
+ logger.error("Have error: {}", session, cause);
+ }
+
+ @Override
+ public void messageReceived(IoSession session, Object message) throws Exception {
+ if (message instanceof IMessage) {
+ IMessage iMessage = (IMessage) message;
+ onMessageReceived(iMessage);
+ processMessage(serverSession.get(), (IMessage) message);
+ }
+ }
+
+ @Override
+ public void messageSent(IoSession session, Object message) throws Exception {
+ if (message instanceof IMessage && session != null) {
+ IMessage iMessage = (IMessage)message;
+ onMessageSent(iMessage);
+ }
+ }
+
+ /**
+ * Method body is copied from {@link IoHandlerAdapter#inputClosed(IoSession)}
+ */
+ @Override
+ public void inputClosed(IoSession session) {
+ logger.debug("Session input closed: {}", session);
+ session.close(true);
+ }
+
+ protected MINAServerSession createServerSession() {
+ return new MINAServerSession(this);
+ }
+
+ protected MINASession createMINASession(IoSession session) {
+ return new MINASession(serviceName, session, loggingConfigurator);
+ }
+
+ protected abstract Class<? extends AbstractCodec> getCodecClass();
+ protected abstract void configureAcceptor(WrapperNioSocketAcceptor acceptor) throws Exception;
+ protected abstract void onSessionOpen(MINASession session);
+ protected abstract void processMessage(MINAServerSession session, IMessage message) throws Exception;
+ protected abstract void onSessionClosed(MINASession session) throws Exception;
+}

ADD
BackEnd/Core/sailfish-core/src/main/java/com/exactpro/sf/services/mina/AbstractMINATCPServerSettings.java
+/*******************************************************************************
+ * Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+ *
+ * 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.exactpro.sf.services.mina;
+
+import com.exactpro.sf.aml.Description;
+import com.exactpro.sf.configuration.suri.SailfishURI;
+import com.exactpro.sf.services.RequiredParam;
+import com.exactpro.sf.services.codecs.ICodecSettings;
+import com.exactpro.sf.services.util.ServiceUtil;
+
+public class AbstractMINATCPServerSettings extends AbstractMINASettings implements ICodecSettings {
+
+ @Description("Service`s hostname. Example: localhost")
+ @RequiredParam
+ private String host = "localhost";
+
+ @Description("Service`s port. Example: 8888")
+ @RequiredParam
+ private int port = 8888;
+
+ @Description("Fields value for which allow receive messages.\n"
+ + "Example: '1, A' or '"+ ServiceUtil.ALIAS_PREFIX + "dataA'")
+ private String filterValues;
+
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ @Override
+ public SailfishURI getDictionaryURI() {
+ return dictionaryName;
+ }
+
+ @Override
+ public String getFilterValues() {
+ return filterValues;
+ }
+
+ public void setFilterValues(String filterValues) {
+ this.filterValues = filterValues;
+ }
+}

ADD
BackEnd/Core/sailfish-core/src/main/java/com/exactpro/sf/services/mina/MINAServerSession.java
+/*******************************************************************************
+ * Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+ *
+ * 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.exactpro.sf.services.mina;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.mina.core.future.WriteFuture;
+import org.apache.mina.core.session.IoSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.exactpro.sf.common.messages.IMessage;
+import com.exactpro.sf.common.util.SendMessageFailedException;
+import com.exactpro.sf.services.ISession;
+import com.exactpro.sf.services.util.CloseSessionException;
+
+public class MINAServerSession implements ISession {
+
+ private static final int DEFAULT_TIMEOUT = 1000;
+ protected final Logger logger = LoggerFactory.getLogger(getClass().getName() + '@' + Integer.toHexString(hashCode()));
+
+ protected AbstractMINATCPServer server;
+
+ public MINAServerSession(AbstractMINATCPServer server) {
+ this.server = server;
+ logger.info("Server session created: {}", this);
+ }
+
+ @Override
+ public String getName() {
+ return server.getServiceName().toString();
+ }
+
+ protected Object prepareMessage(Object message) {
+ return message;
+ }
+
+ @Override
+ public IMessage send(Object message) throws InterruptedException {
+ return send(prepareMessage(message), DEFAULT_TIMEOUT);
+ }
+
+ public IMessage send(Object message, long timeout) throws InterruptedException {
+ List<WriteFuture> futures = new ArrayList<>();
+ Set<String> errorSending = new HashSet<>();
+
+ for (IoSession session : server.sessions.keySet()) {
+ futures.add(session.write(message));
+ }
+
+ long waitUntil = System.currentTimeMillis() + timeout;
+ for (WriteFuture future : futures) {
+ future.await(waitUntil - System.currentTimeMillis());
+ if (!future.isDone() || !future.isWritten()) {
+ errorSending.add(future.getSession().toString());
+ }
+ }
+
+ if (!errorSending.isEmpty()) {
+ throw new SendMessageFailedException(String.format("Message wasn't send during %d milliseconds. %s", timeout, String.join(System.lineSeparator(), errorSending)));
+ }
+
+ return message instanceof IMessage ? (IMessage) message : null;
+ }
+
+ @Override
+ public IMessage sendDirty(Object message) {
+ throw new UnsupportedOperationException("Dirty send is not supported");
+ }
+
+ @Override
+ public void close() {
+ Set<String> errorClosing = new HashSet<>();
+
+ server.sessions.forEach((ioSession, minaSession) -> {
+ try {
+ minaSession.close();
+ } catch (Exception e) {
+ errorClosing.add(ioSession.toString());
+ }
+ });
+
+ if (!errorClosing.isEmpty()) {
+ throw new CloseSessionException("Sessions send error on close." + String.join(System.lineSeparator(), errorClosing));
+ }
+ }
+
+ public boolean isConnected() {
+ return !isClosed();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return server.sessions.isEmpty() || server.sessions.values().stream().allMatch(MINASession::isClosed);
+ }
+
+ @Override
+ public boolean isLoggedOn() {
+ return server.sessions.values().stream().anyMatch(MINASession::isLoggedOn);
+ }
+}

ADD
BackEnd/Core/sailfish-core/src/main/java/com/exactpro/sf/services/util/CloseSessionException.java
+/*******************************************************************************
+ * Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+ *
+ * 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.exactpro.sf.services.util;
+
+import com.exactpro.sf.services.ServiceException;
+
+public class CloseSessionException extends ServiceException {
+ private static final long serialVersionUID = -1336216893108240799L;
+
+ public CloseSessionException() {}
+
+ public CloseSessionException(String message) {
+ super(message);
+ }
+
+ public CloseSessionException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public CloseSessionException(Throwable cause) {
+ super(cause);
+ }
+}

MODIFY
BackEnd/Service/service-itch/src/main/java/com/exactpro/sf/services/itch/ITCHCodecSettings.java
BackEnd/Service/service-itch/src/main/java/com/exactpro/sf/services/itch/ITCHCodecSettings.java
******************************************************************************/
package com.exactpro.sf.services.itch;
-import com.exactpro.sf.common.util.ICommonSettings;
-import com.exactpro.sf.configuration.suri.SailfishURI;
import org.apache.commons.configuration.HierarchicalConfiguration;
-public class ITCHCodecSettings implements ICommonSettings {
+import com.exactpro.sf.configuration.suri.SailfishURI;
+import com.exactpro.sf.services.codecs.ICodecSettings;
+
+public class ITCHCodecSettings implements ICodecSettings {
private byte[] chunkDelimiter;
@@ -59,6 +60,7 @@ public class ITCHCodecSettings implements ICommonSettings {
this.chunkDelimiter = chunkDelimiter;
}
+ @Override
public String getFilterValues() {
return filterValues;
}
@@ -67,6 +69,7 @@ public class ITCHCodecSettings implements ICommonSettings {
this.filterValues = filterValues;
}
+ @Override
public SailfishURI getDictionaryURI() {
return dictionaryURI;
}

[RM] Feature #61754. Create abstract classes for mina server.

Change-Id: Ib67cee9640f0715988e22845e48e32212a9b8e08