<commit_before>/*
 *
 * 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.orientechnologies.orient.server.handler.distributed;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicInteger;

import javax.crypto.SecretKey;

import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.orient.core.command.OCommandOutputListener;
import com.orientechnologies.orient.core.config.OContextConfiguration;
import com.orientechnologies.orient.core.config.OGlobalConfiguration;
import com.orientechnologies.orient.core.db.record.ODatabaseRecord;
import com.orientechnologies.orient.core.db.tool.ODatabaseExport;
import com.orientechnologies.orient.core.record.ORecordInternal;
import com.orientechnologies.orient.core.tx.OTransactionEntry;
import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryClient;
import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryOutputStream;
import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryProtocol;
import com.orientechnologies.orient.enterprise.channel.distributed.OChannelDistributedProtocol;

/**
 * Contains all the information about a cluster node.
 * 
 * @author <NAME> (l.garulli--at--orient<EMAIL>.<EMAIL>)
 * 
 */
public class ODistributedServerNodeRemote implements OCommandOutputListener {
	public enum STATUS {
		DISCONNECTED, CONNECTING, CONNECTED, UNREACHABLE, SYNCHRONIZING
	}

	public enum SYNCH_TYPE {
		SYNCHRONOUS, ASYNCHRONOUS
	}

	private String																id;
	public String																	networkAddress;
	public int																		networkPort;
	public Date																		joinedOn;
	private ODistributedServerManager							manager;
	private OChannelBinaryClient									channel;
	private OContextConfiguration									configuration;
	private volatile STATUS												status					= STATUS.DISCONNECTED;
	private List<OTransactionEntry>								bufferedChanges	= new ArrayList<OTransactionEntry>();
	private int																		clientTxId;
	private long																	lastHeartBeat		= 0;
	private static AtomicInteger									serialClientId	= new AtomicInteger(-1);
	private final ExecutorService									asynchExecutor;
	private Map<String, OServerNodeDatabaseEntry>	databases				= new HashMap<String, OServerNodeDatabaseEntry>();

	public ODistributedServerNodeRemote(final ODistributedServerManager iNode, final String iServerAddress, final int iServerPort) {
		manager = iNode;
		networkAddress = iServerAddress;
		networkPort = iServerPort;
		joinedOn = new Date();
		configuration = new OContextConfiguration();
		id = networkAddress + ":" + networkPort;
		status = STATUS.CONNECTING;

		asynchExecutor = Executors.newSingleThreadExecutor();
	}

	public void connect(final int iTimeout, final String iClusterName, final SecretKey iSecurityKey) throws IOException {
		configuration.setValue(OGlobalConfiguration.NETWORK_SOCKET_TIMEOUT, iTimeout);
		channel = new OChannelBinaryClient(networkAddress, networkPort, configuration);

		OChannelBinaryProtocol.checkProtocolVersion(channel);

		OLogManager.instance().warn(this, "Joining the server node %s:%d to the cluster...", networkAddress, networkPort);

		clientTxId = serialClientId.decrementAndGet();

		// CONNECT TO THE SERVER
		channel.writeByte(OChannelDistributedProtocol.REQUEST_DISTRIBUTED_CONNECT);
		channel.writeString(iClusterName);
		channel.writeBytes(iSecurityKey.getEncoded());
		channel.writeInt(clientTxId);
		channel.flush();
		channel.readStatus();

		// CONNECT EACH DATABASES
		final List<OServerNodeDatabaseEntry> servers = new ArrayList<OServerNodeDatabaseEntry>(databases.values());
		for (OServerNodeDatabaseEntry entry : servers) {
			channel.writeByte(OChannelDistributedProtocol.REQUEST_DISTRIBUTED_DB_OPEN);
			channel.writeInt(clientTxId);

			// PACKET DB INFO TO SEND
			channel.writeString(entry.databaseName);
			channel.writeString(entry.userName);
			channel.writeString(entry.userPassword);
			channel.flush();

			channel.readStatus();
			entry.sessionId = channel.readInt();

			// GET AND CHECK VERSION IF IT'S THE SAME
			final long version = channel.readLong();
			if (version != entry.version) {
				OLogManager.instance().warn(
						this,
						"Remote database '" + entry.databaseName + "' has different version than Leader node (" + entry.version
								+ ") and remote (" + version + "). Removing database from shared list.");

				databases.remove(entry.databaseName);

				// throw new ODistributedException("Database version doesn't match between current node (" + entry.version +<commit_msg>Fixed security check in cluster configuration<commit_after>/*
 *
 * 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.orientechnologies.orient.server.handler.distributed;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicInteger;

import javax.crypto.SecretKey;

import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.orient.core.command.OCommandOutputListener;
import com.orientechnologies.orient.core.config.OContextConfiguration;
import com.orientechnologies.orient.core.config.OGlobalConfiguration;
import com.orientechnologies.orient.core.db.record.ODatabaseRecord;
import com.orientechnologies.orient.core.db.tool.ODatabaseExport;
import com.orientechnologies.orient.core.record.ORecordInternal;
import com.orientechnologies.orient.core.tx.OTransactionEntry;
import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryClient;
import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryOutputStream;
import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryProtocol;
import com.orientechnologies.orient.enterprise.channel.distributed.OChannelDistributedProtocol;

/**
 * Contains all the information about a cluster node.
 * 
 * @author <NAME> (l.garulli--at--orientechnologies.com)
 * 
 */
public class ODistributedServerNodeRemote implements OCommandOutputListener {
	public enum STATUS {
		DISCONNECTED, CONNECTING, CONNECTED, UNREACHABLE, SYNCHRONIZING
	}

	public enum SYNCH_TYPE {
		SYNCHRONOUS, ASYNCHRONOUS
	}

	private String																id;
	public String																	networkAddress;
	public int																		networkPort;
	public Date																		joinedOn;
	private ODistributedServerManager							manager;
	private OChannelBinaryClient									channel;
	private OContextConfiguration									configuration;
	private volatile STATUS												status					= STATUS.DISCONNECTED;
	private List<OTransactionEntry>								bufferedChanges	= new ArrayList<OTransactionEntry>();
	private int																		clientTxId;
	private long																	lastHeartBeat		= 0;
	private static AtomicInteger									serialClientId	= new AtomicInteger(-1);
	private final ExecutorService									asynchExecutor;
	private Map<String, OServerNodeDatabaseEntry>	databases				= new HashMap<String, OServerNodeDatabaseEntry>();

	public ODistributedServerNodeRemote(final ODistributedServerManager iNode, final String iServerAddress, final int iServerPort) {
		manager = iNode;
		networkAddress = iServerAddress;
		networkPort = iServerPort;
		joinedOn = new Date();
		configuration = new OContextConfiguration();
		id = networkAddress + ":" + networkPort;
		status = STATUS.CONNECTING;

		asynchExecutor = Executors.newSingleThreadExecutor();
	}

	public void connect(final int iTimeout, final String iClusterName, final SecretKey iSecurityKey) throws IOException {
		configuration.setValue(OGlobalConfiguration.NETWORK_SOCKET_TIMEOUT, iTimeout);
		channel = new OChannelBinaryClient(networkAddress, networkPort, configuration);

		OChannelBinaryProtocol.checkProtocolVersion(channel);

		OLogManager.instance().warn(this, "Joining the server node %s:%d to the cluster...", networkAddress, networkPort);

		clientTxId = serialClientId.decrementAndGet();

		// CONNECT TO THE SERVER
		channel.writeByte(OChannelDistributedProtocol.REQUEST_DISTRIBUTED_CONNECT);
		channel.writeInt(clientTxId);

		channel.writeString(iClusterName);
		channel.writeBytes(iSecurityKey.getEncoded());

		channel.flush();
		channel.readStatus();

		// CONNECT EACH DATABASES
		final List<OServerNodeDatabaseEntry> servers = new ArrayList<OServerNodeDatabaseEntry>(databases.values());
		for (OServerNodeDatabaseEntry entry : servers) {
			channel.writeByte(OChannelDistributedProtocol.REQUEST_DISTRIBUTED_DB_OPEN);
			channel.writeInt(clientTxId);

			// PACKET DB INFO TO SEND
			channel.writeString(entry.databaseName);
			channel.writeString(entry.userName);
			channel.writeString(entry.userPassword);
			channel.flush();

			channel.readStatus();
			entry.sessionId = channel.readInt();

			// GET AND CHECK VERSION IF IT'S THE SAME
			final long version = channel.readLong();
			if (version != entry.version) {
				OLogManager.instance().warn(
						this,
						"Remote database '" + entry.databaseName + "' has different version than Leader node (" + entry.version
								+ ") and remote (" + version + "). Removing database from shared list.");

				databases.remove(entry.databaseName);

				// throw new ODistributedException("Database version doesn't match between current node (" + entry.version +
{'fraction_non_alphanumeric': 0.08636701337295691, 'fraction_numerical': 0.001114413075780089, 'mean_word_length': 4.6323221757322175, 'pattern_counts': {'":': 2, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1925099', 'n_tokens_mistral': 3444, 'n_tokens_neox': 2961, 'n_words': 721}