public override DbDataReader ExecuteReader(string sql, CommandType cmdType, CommandBehavior cmdBehavior, params DbParameter[] cmdParams)
 {
     DbConnection conn = this.CreateConnection();
     DbCommand cmd = this.CreateCommand(sql, cmdType, cmdParams);
     this.PrepareCmd(cmd, conn);
     return this.ExecuteReader(cmd, cmdBehavior);
 }

        public static void ExecuteReader(SqlConnection connectionRef, string procedureName, SqlParameter[] parameters, CommandBehavior behavior, Action<SqlDataReader> readerCallback)
        {
            using (SqlCommand command = new SqlCommand(procedureName, connectionRef))
            {
                command.CommandType = CommandType.StoredProcedure;
                command.CommandTimeout = 180;

                if (parameters.Length > 0)
                {
                    foreach (SqlParameter param in parameters)
                    {
                        command.Parameters.Add(param);
                    }
                }

                SqlDataReader reader = command.ExecuteReader(behavior);

                try
                {
                    readerCallback(reader);
                }
                finally
                {
                    if (reader != null)
                        reader.Close();
                }
            }
        }

		public IDataReader ExecuteReader(CommandBehavior behavior)
		{
			var reliableConnection = (ReliableSqlConnection) this.Connection;
			return this.command.ExecuteReaderWithRetry(behavior,
			                                           reliableConnection.CommandRetryPolicy,
			                                           reliableConnection.ConnectionRetryPolicy);
		}

        public SqlCommand GetDBCommand(SqlConnection sqlcn, String CmdText, CommandType CmdType, CommandBehavior CmdBehavior, SqlParameter[] sqlParam)
        {
            SqlCommand sqlcmd = null;

            try
            {
                sqlcmd = new SqlCommand(CmdText, sqlcn);
                sqlcmd.CommandType = CmdType;

                sqlcmd.CommandTimeout = GetCommandTimeout();

                Utilities.DebugLogging.Log("CONNECTION STRING " + sqlcn.ConnectionString);
                Utilities.DebugLogging.Log("COMMAND TEXT " + CmdText);
                Utilities.DebugLogging.Log("COMMAND TYPE " + CmdType.ToString());
                if (sqlParam != null)
                    Utilities.DebugLogging.Log("NUMBER OF PARAMS " + sqlParam.Length);

                AddSQLParameters(sqlcmd, sqlParam);

            }
            catch (Exception ex)
            {
                Utilities.DebugLogging.Log(ex.Message);
                Utilities.DebugLogging.Log(ex.StackTrace);
            }
            return sqlcmd;
        }

 internal OdbcDataReader(OdbcCommand command, CMDWrapper cmdWrapper, CommandBehavior commandbehavior)
 {
     this.command = command;
     this._commandBehavior = commandbehavior;
     this._cmdText = command.CommandText;
     this._cmdWrapper = cmdWrapper;
 }

 internal OracleDataReader(OracleCommand command, ArrayList refCursorParameterOrdinals, string statementText, CommandBehavior commandBehavior)
 {
     this.ObjectID = Interlocked.Increment(ref _objectTypeCount);
     this._commandBehavior = commandBehavior;
     this._statementText = statementText;
     this._closeConnectionToo = this.IsCommandBehavior(CommandBehavior.CloseConnection);
     if (CommandType.Text == command.CommandType)
     {
         this._keyInfoRequested = this.IsCommandBehavior(CommandBehavior.KeyInfo);
     }
     ArrayList list = new ArrayList();
     int num2 = 0;
     OracleDataReader reader = null;
     for (int i = 0; i < refCursorParameterOrdinals.Count; i++)
     {
         int num3 = (int) refCursorParameterOrdinals[i];
         OracleParameter parameter = command.Parameters[num3];
         if (OracleType.Cursor == parameter.OracleType)
         {
             reader = (OracleDataReader) parameter.Value;
             reader._recordsAffected = num2;
             list.Add(reader);
             parameter.Value = DBNull.Value;
         }
         else
         {
             num2 += (int) parameter.Value;
         }
     }
     this._refCursorDataReaders = new OracleDataReader[list.Count];
     list.CopyTo(this._refCursorDataReaders);
     this._nextRefCursor = 0;
     this.NextResultInternal();
 }

    /// <summary>
    /// Initiates the asynchronous execution of the Transact-SQL statement or stored procedure using one of the 
    /// <see cref="CommandBehavior"/> values, and retrieving one or more result sets from the server.
    /// </summary>
    /// <param name="command">The <see cref="SqlCommand"/> to be executed.</param>
    /// <param name="behavior">One of the <see cref="CommandBehavior"/> values.</param>
    /// <returns>A singleton observable sequence containing a <see cref="SqlDataReader"/> object that provides access to
    /// the result sets of the specified <paramref name="command"/>.</returns>
    public static IObservable<SqlDataReader> ExecuteReaderObservable(this SqlCommand command, CommandBehavior behavior)
    {
      Contract.Requires(command != null);
      Contract.Ensures(Contract.Result<IObservable<SqlDataReader>>() != null);

      return Observable.StartAsync(cancel => command.ExecuteReaderAsync(behavior, cancel));
    }

 public static Task<SqlDataReader> ExecuteReaderAsyncWithRetry(
                                                               SqlCommand command,
                                                               CommandBehavior behavior,
                                                               IRetryPolicy retryPolicy)
 {
     return retryPolicy.ExecuteAsyncWithRetry(() => command.ExecuteReaderAsync(behavior));
 }

        /// Runs a stored procedure
        /// </summary>
        /// <param name="procedureName"></param>
        /// <param name="QueryParameters"></param>
        /// <param name="OutputParameters"></param>
        /// <param name="DB"></param>
        public virtual void RunStoredProcedure(IDbConnection connection, string procedureName,
            IEnumerable<IDataParameter> inputParameters,
            IEnumerable<IDataParameter> outputParameters,
            IDbTransaction transaction,
            CommandBehavior commandBehavior)
        {
            SqlCommand cmd = new SqlCommand(procedureName, (SqlConnection)connection);
            cmd.CommandType = CommandType.StoredProcedure;

            ProcessSql(procedureName, inputParameters);

            foreach (var parm in inputParameters)
            {
                cmd.Parameters.Add(parm);
            }
            if (outputParameters != null)
            {
                foreach (SqlParameter parm in outputParameters)
                {
                    parm.Direction = ParameterDirection.Output;
                    //OutputParameters[i].Value = -1;
                    cmd.Parameters.Add(parm);
                }
            }

            ExecuteSqlFinal(new Action(() =>
            {
                cmd.ExecuteScalar();
            }));
            if (commandBehavior == CommandBehavior.CloseConnection)
            {
                connection.Close();
            }
            OnQueryComplete();
        }

 public IDataReader ExecuteReader(CommandBehavior behavior)
 {
     using (ExecuteHelper.Begin(dur => context.FireExecuteEvent(this, string.Format("ExecuteReader({0})", behavior), dur)))
     {
         return command.ExecuteReader(behavior);
     }
 }

        public IDataReader ExecuteDataReader(Saga.Data.IQueryProvider query, CommandBehavior behavior)
        {
            MySqlConnection connection = ConnectionPool.Request();
            MySqlCommand command = new MySqlCommand();
            MySqlDataReader reader = null;

            try
            {

                command.CommandText = query.CmdText;
                command.Connection = connection;
                foreach (KeyValuePair<string, object> pair in query.Parameters)
                {
                    command.Parameters.AddWithValue(pair.Key, pair.Value);
                }

                return command.ExecuteReader(behavior);
            }
            catch (Exception e)
            {
                __dbtracelog.WriteError("Database", e.Message);
                return null;
            }
            finally
            {
                //ALWAYS CLOSE THE CONNECTION AND REPOOL THE ITEMS
                if (reader != null && reader.IsClosed == false) reader.Close();
                ConnectionPool.Release(connection);
            }
        }

 public static DbDataReader ExecuteProfilableReader(this DbCommand dbCommand, CommandBehavior commandBehavior = CommandBehavior.Default)
 {
     if (!(dbCommand is ProfilableDbCommand))
         return (new ProfilableDbCommand(dbCommand, dbCommand.Connection)).ExecuteReader(commandBehavior);
     else
         return dbCommand.ExecuteReader(commandBehavior);
 }

 public virtual int Count(IDbConnection connection, string query, IEnumerable<IDataParameter> parameters = null,
     CommandBehavior commandBehavior = CommandBehavior.Default)
 {
     ValidateQueryType(query, "select");
     string countQuery = "SELECT COUNT(*) FROM (" + query + ") q";
     return RunQueryScalar(connection,countQuery, parameters,null,commandBehavior);
 }

        public virtual IDataReader RunQuery(IDbConnection connection, string query, 
            IEnumerable<IDataParameter> parameters = null,
            int firstRow=-1,
            int totalRows=-1,
            IDbTransaction transaction = null,
            CommandBehavior commandBehavior = CommandBehavior.Default)
        {
            IDataReader dataReader = null;
            string sql;
            if (firstRow >= 0 || totalRows >= 0)
            {
                sql = GetQueryForRows(query, firstRow, totalRows);
            } else {
                sql = query;
            }
            sql = ProcessSql(sql,parameters);

            using (IDbCommand cmd = GetCommand(connection, sql, parameters, transaction))
            {

                ExecuteSqlFinal(new Action(() =>
                {
                    dataReader = cmd.ExecuteReader(commandBehavior);
                }));

                cmd.Parameters.Clear();
            }

            OnQueryComplete();
            return dataReader;
        }

    public SqlCommand GetDBCommand(SqlConnection sqlcn, String CmdText, CommandType CmdType, CommandBehavior CmdBehavior, SqlParameter[] sqlParam)
    {
        SqlCommand sqlcmd = new SqlCommand(CmdText, sqlcn);
            sqlcmd.CommandType = CmdType;

            AddSQLParameters(sqlcmd, sqlParam);
            return sqlcmd;
    }

 protected override DbDataReader ExecuteDbDataReader(CommandBehavior behavior)
 {
     string query, verb, url;
     ParseCommantText(out query, out verb, out url);
     var json = ElasticDbConnection.ExecuteRequest(verb, url, query);
     var reader = new ElasticDbDataReader(json, ElasticDbConnection.ResultParserTypes);
     return reader;
 }

 public virtual SqlDataReader RunStoredProcedureDataset(IDbConnection connection, 
     string procedureName,
     IEnumerable<IDataParameter> queryParameters,
     IDbTransaction transaction,
     CommandBehavior commandBehavior)
 {
     return RunStoredProcedureDataset(connection,procedureName, queryParameters, null,transaction, commandBehavior);
 }

 public static Task<SqlDataReader> ExecuteReaderSyncOrAsync(this SqlCommand command, CommandBehavior cb, CancellationToken token, Random rnd)
 {
     return AsyncUtils.SyncOrAsyncMethod(
         () => command.ExecuteReader(cb),
         () => command.ExecuteReaderAsync(cb, token),
         AsyncUtils.ChooseSyncAsyncMode(rnd)
         );
 }

 public override IAsyncResult BeginExecuteReader(AsyncCallback callback, object state, CommandBehavior behavior)
 {
     if (IsTraced)
     {
         Tracer.Instance.EnterFunction(TraceClass, "AsyncExecuteReader", new string[] { "CommandText", "callback", "state", "behavior", "Parameters" }, new object[] { this.CommandText, callback, state, behavior, base.GetTraceParameters(true) });
     }
     return ((SqlCommand) base.DbCommand).BeginExecuteReader(callback, state, behavior);
 }

		public MockDataReader(MockDataCommand command, CommandBehavior behavior)
			: base()
		{
			if (command == null) throw new ArgumentNullException(nameof(command));

			this.Command = command;
			this.CommandBehavior = behavior;
		}