<commit_before>// <copyright file="RiakPbcSocket.cs" company="Basho Technologies, Inc.">
// Copyright 2011 - <NAME> & <NAME>
// Copyright 2014 - Basho Technologies, Inc.
//
// This file is provided to you 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.
// </copyright>

namespace RiakClient.Comms
{
    using System;
    using System.IO;
    using System.Net;
    using System.Net.Security;
    using System.Net.Sockets;
    using System.Security.Authentication;
    using Auth;
    using Commands;
    using Config;
    using Exceptions;
    using Extensions;
    using Messages;
    using ProtoBuf;

    internal class RiakPbcSocket : IDisposable
    {
        private const int SizeOfInt = sizeof(int);
        private const int PbMsgHeaderSize = SizeOfInt + sizeof(byte);

        private readonly string server;
        private readonly int port;
        private readonly Timeout connectTimeout;
        private readonly Timeout readTimeout;
        private readonly Timeout writeTimeout;
        private readonly RiakSecurityManager securityManager;
        private readonly bool checkCertificateRevocation = false;

        private Stream networkStream = null;

        public RiakPbcSocket(IRiakNodeConfiguration nodeConfig, IRiakAuthenticationConfiguration authConfig)
        {
            server = nodeConfig.HostAddress;
            port = nodeConfig.PbcPort;
            readTimeout = nodeConfig.NetworkReadTimeout;
            writeTimeout = nodeConfig.NetworkWriteTimeout;
            connectTimeout = nodeConfig.NetworkConnectTimeout;
            securityManager = new RiakSecurityManager(server, authConfig);
            checkCertificateRevocation = authConfig.CheckCertificateRevocation;
        }

        private Stream NetworkStream
        {
            get
            {
                if (networkStream == null)
                {
                    SetUpNetworkStream();
                }

                return networkStream;
            }
        }

        public void Write(MessageCode messageCode)
        {
            var messageBody = new byte[PbMsgHeaderSize];

            var size = BitConverter.GetBytes(IPAddress.HostToNetworkOrder(1));
            Array.Copy(size, messageBody, SizeOfInt);
            messageBody[SizeOfInt] = (byte)messageCode;

            NetworkStream.Write(messageBody, 0, PbMsgHeaderSize);
        }

        public RiakResult Write(IRiakCommand command)
        {
            RpbReq request = command.ConstructPbRequest();
            if (request.IsMessageCodeOnly)
            {
                Write(request.MessageCode);
                return RiakResult.Success();
            }
            else
            {
                return DoWrite(s => Serializer.Serialize(s, request), request.GetType());
            }
        }

        public void Write<T>(T message) where T : class
        {
            DoWrite(s => Serializer.Serialize(s, message), typeof(T));
        }

        public MessageCode Read(MessageCode expectedCode)
        {
            int size = 0;
            MessageCode messageCode = ReceiveHeader(out size);

            if (expectedCode != messageCode)
            {
                string errorMessage = string.Format("Expected return code {0} received {1}", expectedCode, messageCode);
                throw new RiakException(errorMessage, false);
            }

            return messageCode;
        }

        public RiakResult Read(IRiakCommand command)
        {
            MessageCode expectedCode = command.ExpectedCode;
            Type expectedType = MessageCodeTypeMapBuilder.GetTypeFor(expectedCode);

            int size = DoRead(expectedCode, expectedType);

            RpbResp response = DeserializeInstance(expectedType, size);
            command.OnSuccess(response);

            return RiakResult.Success();
        }

        public T Read<T>() where T : new()
        {
            Type expectedType = typeof(T);
            MessageCode expectedCode = MessageCodeTypeMapBuilder.GetMessageCodeFor(expectedType);

            int size = DoRead(expectedCode, expectedType);
            return DeserializeInstance<T>(size);
        }

        public void Dispose()
        {
            Disconnect();
        }

        public void Disconnect()
        {
            // NB: since networkStream owns the underlying socket there is no need to close socket as well
            if (networkStream != null)
            {
                networkStream.Close();
                networkStream.Dispose();
            }<commit_msg>Fix - Creating new NetworkStreamwhen previous one is disconnected<commit_after>// <copyright file="RiakPbcSocket.cs" company="Basho Technologies, Inc.">
// Copyright 2011 - <NAME> & <NAME>
// Copyright 2014 - Basho Technologies, Inc.
//
// This file is provided to you 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.
// </copyright>

namespace RiakClient.Comms
{
    using System;
    using System.IO;
    using System.Net;
    using System.Net.Security;
    using System.Net.Sockets;
    using System.Security.Authentication;
    using Auth;
    using Commands;
    using Config;
    using Exceptions;
    using Extensions;
    using Messages;
    using ProtoBuf;

    internal class RiakPbcSocket : IDisposable
    {
        private const int SizeOfInt = sizeof(int);
        private const int PbMsgHeaderSize = SizeOfInt + sizeof(byte);

        private readonly string server;
        private readonly int port;
        private readonly Timeout connectTimeout;
        private readonly Timeout readTimeout;
        private readonly Timeout writeTimeout;
        private readonly RiakSecurityManager securityManager;
        private readonly bool checkCertificateRevocation = false;

        private Stream networkStream = null;

        public RiakPbcSocket(IRiakNodeConfiguration nodeConfig, IRiakAuthenticationConfiguration authConfig)
        {
            server = nodeConfig.HostAddress;
            port = nodeConfig.PbcPort;
            readTimeout = nodeConfig.NetworkReadTimeout;
            writeTimeout = nodeConfig.NetworkWriteTimeout;
            connectTimeout = nodeConfig.NetworkConnectTimeout;
            securityManager = new RiakSecurityManager(server, authConfig);
            checkCertificateRevocation = authConfig.CheckCertificateRevocation;
        }

        private Stream NetworkStream
        {
            get
            {
                if (networkStream == null)
                {
                    SetUpNetworkStream();
                }

                return networkStream;
            }
        }

        public void Write(MessageCode messageCode)
        {
            var messageBody = new byte[PbMsgHeaderSize];

            var size = BitConverter.GetBytes(IPAddress.HostToNetworkOrder(1));
            Array.Copy(size, messageBody, SizeOfInt);
            messageBody[SizeOfInt] = (byte)messageCode;

            NetworkStream.Write(messageBody, 0, PbMsgHeaderSize);
        }

        public RiakResult Write(IRiakCommand command)
        {
            RpbReq request = command.ConstructPbRequest();
            if (request.IsMessageCodeOnly)
            {
                Write(request.MessageCode);
                return RiakResult.Success();
            }
            else
            {
                return DoWrite(s => Serializer.Serialize(s, request), request.GetType());
            }
        }

        public void Write<T>(T message) where T : class
        {
            DoWrite(s => Serializer.Serialize(s, message), typeof(T));
        }

        public MessageCode Read(MessageCode expectedCode)
        {
            int size = 0;
            MessageCode messageCode = ReceiveHeader(out size);

            if (expectedCode != messageCode)
            {
                string errorMessage = string.Format("Expected return code {0} received {1}", expectedCode, messageCode);
                throw new RiakException(errorMessage, false);
            }

            return messageCode;
        }

        public RiakResult Read(IRiakCommand command)
        {
            MessageCode expectedCode = command.ExpectedCode;
            Type expectedType = MessageCodeTypeMapBuilder.GetTypeFor(expectedCode);

            int size = DoRead(expectedCode, expectedType);

            RpbResp response = DeserializeInstance(expectedType, size);
            command.OnSuccess(response);

            return RiakResult.Success();
        }

        public T Read<T>() where T : new()
        {
            Type expectedType = typeof(T);
            MessageCode expectedCode = MessageCodeTypeMapBuilder.GetMessageCodeFor(expectedType);

            int size = DoRead(expectedCode, expectedType);
            return DeserializeInstance<T>(size);
        }

        public void Dispose()
        {
            Disconnect();
        }

        public void Disconnect()
        {
            // NB: since networkStream owns the underlying socket there is no need to close socket as well
            if (networkStream != null)
            {
                networkStream.Close();
                networkStream.Dispose();
                networkStream = null;
            }
{'fraction_non_alphanumeric': 0.07529527559055119, 'fraction_numerical': 0.0033464566929133858, 'mean_word_length': 2.0504353047132993, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, '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': '3089431', 'n_tokens_mistral': 2515, 'n_tokens_neox': 2384, 'n_words': 711}