unit MySQLConnection;

// Copyright (C) 2003, 2004 MySQL AB
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

interface

uses
  gnugettext, Windows, Messages, SysUtils, Classes, ComCtrls, ImgList,
  Controls, TntComCtrls, TntClasses, Forms, Contnrs,
  ExtCtrls, AuxFuncs, PNGImage, StdCtrls,
  SyncObjs, myx_public_interface, MyxError, Options;

const
  WM_Disconnected = WM_USER + 101;
  WM_Reconnected = WM_USER + 102;
  WM_DefaultSchemaChanged = WM_USER + 103;
  WM_SchemaListChanged = WM_USER + 104;

  CR_SERVER_GONE_ERROR = 2006;
  CR_SERVER_LOST = 2013;
  CR_CONN_HOST_ERROR = 2003;

type
  TFetchDataThread = class;

  DataBeingFetchedSet = set of 1..100;

  TThreadExecMethod = procedure(Sender: TObject) of object;

  IMySQLConnTransactionStatusChangeListener = interface
    procedure TransactionStatusChanged;
    procedure TransactionQueryExecuted(SQLCmd: WideString);
    procedure TransactionStarted;
  end;

  TMySQLConn = class(TObject)
  private
    FMySQL: Pointer;                             // Main MySQL server connection data.
    FDefaultSchema: WideString;
    FInTransaction: Boolean;
    FTransactionStatusChangeListeners: TList;
    FQuoteChar: WideString;
    FWorkList: TThreadList;
    FStatusBar: TTntStatusBar;

    function ShowConnectToInstanceForm(var User_connection: TMYX_USER_CONNECTION; var MySQL: Pointer;
      MySQLConnectionsOnly: Boolean; SelectSchemata: Boolean): Integer;
    function ConnectUsingUserConnection(Connection: WideString; var User_connection: TMYX_USER_CONNECTION;
      var FMySQL: Pointer; SelectSchemata: Boolean): Integer;
    function ConnectUsingCommandlineConnectionParams(var User_connection: TMYX_USER_CONNECTION; var FMySQL: Pointer;
      SelectSchemata: Boolean): Integer;
    function Connect(var FMySQL: Pointer; user_conn: TMYX_USER_CONNECTION; SelectSchemata: Boolean): Integer;
  protected
    procedure SetDefaultSchema(DefaultSchema: WideString);
    procedure SetInTransaction(InTransaction: Boolean);
    function GetQuoteChar: WideString;
    function GetFetchingData: Boolean;
  public
    MySQLMajorVersion,
    MySQLMinorVersion: Integer;
    user_connection: TMYX_USER_CONNECTION;
    ConnectionCaption: WideString;
    Params: TTntStringList;
    DoParameterSubstitution: Boolean;

    FCurrentDataThread: TFetchDataThread;        // Set by each thread as soon as it starts working. Access is serialized
                                                 // via the critical section.

    FetchDataThreadCriticalSection: TCriticalSection;
    FetchDataMultiReadSync: TMultiReadExclusiveWriteSynchronizer;

    QueryHookMultiReadSync: TMultiReadExclusiveWriteSynchronizer;
    QueryHookSQL: WideString;

    DataBeingFetched: DataBeingFetchedSet;

    Connected: Boolean;
    ConnectedToLocalhost: Boolean;

    constructor Create(StatusBar: TTntStatusBar);
    destructor Destroy; override;

    procedure ClearWorkList;
    procedure RefreshConnectionCaption;
    procedure CheckConnectionStatusChange(SQLCmd: WideString);
    procedure TransactionQueryExecuted;
    function ConnectToServer(AutoConnect: Boolean; MySQLConnectionsOnly: Boolean = True; SelectSchemata:
      Boolean = False): Integer;
    function Reconnect: Boolean;
    procedure Disconnect;
    procedure FetchData(KindOfData: Integer; FetchMethod: TThreadExecMethod; RefreshMethod: TThreadExecMethod;
      Target: TObject; StatusBarText: WideString; AllowParallelExecution: Boolean = False;
      AllowDuplicatedKindOfData: Boolean = False; ControlThread: TThread = nil);
    procedure StopCurrentThread(Timeout: Integer);

    function ExecuteDirect(SQLCmd: WideString; DisplayErrorDialog: Boolean = True): Boolean;

    procedure SchemaListChanged;

    procedure AddTransactionStatusListener(Listener: IMySQLConnTransactionStatusChangeListener);
    procedure RemoveTransactionStatusListener(Listener: IMySQLConnTransactionStatusChangeListener);

    property DefaultSchema: WideString read FDefaultSchema write SetDefaultSchema;
    property InTransaction: Boolean read FInTransaction write SetInTransaction;
    property MySQL: Pointer read FMySQL; 
    property QuoteChar: WideString read GetQuoteChar;
    property FetchingData: Boolean read GetFetchingData;
  end;

  TFetchDataThread = class(TThread)
  private
    FConnection: TMySQLConn;
    ErrorMsg: WideString;
    FetchMethod: TThreadExecMethod;
    RefreshMethod: TThreadExecMethod;
    SynchronizedMethod: TThreadExecMethod;
    KindOfData: Integer;
    FTarget: TObject;
    FStatusBarText: WideString;

    AllowParallelExecution: Boolean;
  protected
    procedure Execute; override;
    procedure SetStatusBar(StatusBarText: WideString);
  public
    constructor Create(AMySQLConn: TMySQLConn; AKindOfData: Integer; AFetchMethod: TThreadExecMethod;
      ARefreshMethod: TThreadExecMethod; ATarget: TObject; AStatusBarText: WideString; AAllowParallelExecution: Boolean);
    destructor Destroy; override;
    procedure ReleaseCriticalSection;
    procedure UpdateStatusBar;
    procedure ShowError;
    procedure ExecuteSynchronized(SynchronizedMethod: TThreadExecMethod);
    procedure ExecuteSynchronizedMethod;

    property Connection: TMySQLConn read FConnection;
    property Target: TObject read FTarget;
    property StatusBarText: WideString write SetStatusBar;
  end;

const
  //KindOfData
  KindOfData_CatalogSchema = 1;
  KindOfData_UserNames = 2;
  KindOfData_ProcessList = 3;
  KindOfData_SchemaTables = 4;
  KindOfData_StatusVars = 5;
  KindOfData_ServerVars = 6;
  KindOfData_UserData = 7;
  KindOfData_SchemaTableStatus = 8;
  KindOfData_TableAction = 9;
  KindOfData_SchemaIndices = 10;

//----------------------------------------------------------------------------------------------------------------------

implementation

uses
  ConnectToInstance;

//----------------------------------------------------------------------------------------------------------------------

constructor TMySQLConn.Create(StatusBar: TTntStatusBar);

begin
  // TODO: DLL Version check still missing
  inherited Create;

  FStatusBar := StatusBar;

  FWorkList := TThreadList.Create;
  FMySQL := nil;
  user_connection := nil;
  FTransactionStatusChangeListeners := nil;

  FDefaultSchema := '';

  Connected := False;
  ConnectedToLocalhost := False;

  //Initialize components for thread handling
  FetchDataThreadCriticalSection := TCriticalSection.Create;
  FetchDataMultiReadSync := TMultiReadExclusiveWriteSynchronizer.Create;
  DataBeingFetched := [];

  Params := TTntStringList.Create;
  DoParameterSubstitution := False;

  QueryHookMultiReadSync := TMultiReadExclusiveWriteSynchronizer.Create;
end;

//----------------------------------------------------------------------------------------------------------------------

destructor TMySQLConn.Destroy;

begin
  ClearWorkList;
  FWorkList.Free;

  FetchDataMultiReadSync.Free;
  FetchDataThreadCriticalSection.Free;

  QueryHookMultiReadSync.Free;

  if (user_connection <> nil) then
    user_connection.Free;

  Params.Free;

  // Close connection and free memory for MySQL struct
  myx_mysql_close(FMySQL);

  inherited;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.ClearWorkList;

// Goes through the work list and tells all waiting threads to terminate. Before that however the current thread must
// be stopped. It is given a timeout of 5 seconds to finish its work otherwise it is killed.

var
  I: Integer;

begin
  with FWorkList.LockList do
  try
    for I := 0 to Count - 1 do
      TFetchDataThread(Items[I]).Terminate;

    // Now that all threads are marked to terminate they will not start any work but immediately return
    // as soon as the fetch lock is free.
    // Now finish the currently active thread if there is one.
    if Assigned(FCurrentDataThread) then
    begin
      FCurrentDataThread.Terminate;

      // Wait 2 seconds and kill the thread if it did not stop working within that time.
      if WaitForSingleObject(FCurrentDataThread.Handle, 2000) = WAIT_TIMEOUT then
        TerminateThread(FCurrentDataThread.Handle, 999);
      FetchDataThreadCriticalSection.Release;
      FreeAndNil(FCurrentDataThread);
    end;
  finally
    FWorkList.UnlockList;
  end;

  repeat
    Sleep(100);
    with FWorkList.LockList do
    try
      I := Count;
    finally
      FWorkList.UnlockList;
    end;
  until I = 0;
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.ConnectToServer(AutoConnect: Boolean; MySQLConnectionsOnly: Boolean = True;
  SelectSchemata: Boolean = False): Integer;
  
begin
  //If AutoConnect and a connection was specified
  if (AutoConnect) and (MYXCommonOptions.ConnectionToUse <> '') then
    Result := ConnectUsingUserConnection(MYXCommonOptions.ConnectionToUse,
      User_connection,
      FMySQL, SelectSchemata)
  //If AutoConnect and username was specified
  else
    if (AutoConnect) and (MYXCommonOptions.ConnectionUsername <> '') then
      Result := ConnectUsingCommandlineConnectionParams(User_connection,
        FMySQL, SelectSchemata)
    else
      Result := ShowConnectToInstanceForm(User_connection, FMySQL, MySQLConnectionsOnly, SelectSchemata);

  if (Result = 1) then
  begin
    MySQLMajorVersion := myx_get_mysql_major_version(FMySQL);
    MySQLMinorVersion := myx_get_mysql_minor_version(FMySQL);

    RefreshConnectionCaption;

    Result := 1;

    Connected := True;
    ConnectedToLocalhost := (myx_is_localhost(user_connection.hostname) = 1);
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.ShowConnectToInstanceForm(var User_connection: TMYX_USER_CONNECTION; var MySQL: Pointer;
  MySQLConnectionsOnly: Boolean; SelectSchemata: Boolean): Integer;

var
  ConnectToInstanceForm: TConnectToInstanceForm;
  ModalRes: Integer;

begin
  ConnectToInstanceForm := TConnectToInstanceForm.Create(nil,
    MySQLConnectionsOnly, SelectSchemata);
  try
    ModalRes := ConnectToInstanceForm.ShowModal;
    if (ModalRes = mrOK) then
    begin
      //Initialize Variables
      User_connection := TMYX_USER_CONNECTION.create(ConnectToInstanceForm.User_Connection.get_record_pointer);
      MySQL := ConnectToInstanceForm.PMySQL;
      DefaultSchema := myx_get_default_schema(FMySQL);

      Result := 1;
    end
    else
      if (ModalRes = mrAbort) then
        Result := -1
      else
        Result := 0;

  finally
    ConnectToInstanceForm.Free;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.ConnectUsingUserConnection(Connection: WideString; var User_connection: TMYX_USER_CONNECTION;
  var FMySQL: Pointer; SelectSchemata: Boolean): Integer;

var
  user_conns: PMYX_USER_CONNECTIONS;
  stored_conns: TMYX_USER_CONNECTIONS;
  user_conn: TMYX_USER_CONNECTION;
  error: MYX_LIB_ERROR;
  i: Integer;
  
begin
  Result := -1;

  //Get connection
  //Fetch connections from library
  user_conns := myx_load_user_connections(
    MYXCommonOptions.UserDataDir + 'mysqlx_user_connections.xml', @error);
  if (error <> MYX_NO_ERROR) then
    raise EMyxCommonLibraryError.Create(_('Error while loading stored connections.'),
      Ord(error), MYXCommonOptions.UserDataDir + 'mysqlx_user_connections.xml');

  try
    stored_conns := TMYX_USER_CONNECTIONS.create(user_conns);
    try
      user_conn := nil;
      for i := 0 to stored_conns.user_connections.Count - 1 do
      begin
        if (CompareText(stored_conns.user_connections[i].connection_name,
          MYXCommonOptions.ConnectionToUse) = 0) then
        begin
          user_conn := stored_conns.user_connections[i];
          break;
        end;
      end;

      //If the connection is found, connect
      if (user_conn <> nil) then
      begin
        //Copy connection
        User_connection := TMYX_USER_CONNECTION.Create(user_conn.get_record_pointer);

        Result := Connect(FMySQL, User_connection, SelectSchemata);
      end
      else
        ShowModalDialog(Application.Title + ' - ' + _('Connection not found'),
          Format(_('The connection %s cannot be found in the list of stored connections.'),
          [MYXCommonOptions.ConnectionToUse]), myx_mtError, _('OK'));
    finally
      stored_conns.Free;
    end;
  finally
    myx_free_user_connections(user_conns);
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.ConnectUsingCommandlineConnectionParams(var User_connection: TMYX_USER_CONNECTION;
  var FMySQL: Pointer; SelectSchemata: Boolean): Integer;
  
begin
  //Create new connection form command line parameters
  User_connection := TMYX_USER_CONNECTION.Create(
    '',
    MYXCommonOptions.ConnectionUsername,
    MYXCommonOptions.ConnectionPassword,
    MYXCommonOptions.ConnectionHost,
    StrToIntDef(MYXCommonOptions.ConnectionPort, 3306),
    MYXCommonOptions.ConnectionSchema,
    '', '', MYX_MYSQL_CONN, MYX_FAVORITE_USER_CONNECTION);

  Result := Connect(FMySQL, User_connection, SelectSchemata);
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.Connect(var FMySQL: Pointer; user_conn: TMYX_USER_CONNECTION; SelectSchemata: Boolean): Integer;

var
  ConnectionResult: Integer;
  
begin
  //Initialize FMySQL
  FMySQL := myx_mysql_init();
  if (FMySQL = nil) then
  begin
    ShowModalDialog(Application.Title + ' ' + _('Error'),
      _('Error while allocating memory for MySQL Struct.'),
      myx_mtError, 'OK');

    Result := -1;
    Exit;
  end;

  //Connect to instance
  ConnectionResult := myx_connect_to_instance(
    user_conn.get_record_pointer,
    FMySQL);
  if (ConnectionResult = 0) then
  begin
    if (SelectSchemata) and (user_conn.schema <> '') then
    begin
      ConnectionResult := myx_use_schema(FMySQL, user_conn.schema);
      if (ConnectionResult <> 0) then
      begin
        ShowModalDialog(Application.Title + ' ' + _('Error'),
          Format(_('The schema %s cannot be selected.'),
          [user_conn.schema]),
          myx_mtError, 'OK');

        Result := -1;
        Exit;
      end;

      DefaultSchema := myx_get_default_schema(FMySQL);

    end;

    Result := 1;
  end
  else
  begin
    ShowModalDialog(Application.Title + ' ' + _('Error'),
      _('Could not connect to the specified host.') + ' ' + #13#10 + #13#10 +
      Format(_('MySQL Error Number %d' + #13#10 + '%s'),
      [myx_mysql_errno(FMySQL),
      myx_mysql_error(FMySQL)]),
        myx_mtError, 'OK');

    Result := -1;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.RefreshConnectionCaption;

begin
  if (User_Connection <> nil) then
  begin
    ConnectionCaption := User_Connection.username + '@' + User_Connection.hostname + ':' + IntToStr(User_Connection.port);
    if FDefaultSchema <> '' then
      ConnectionCaption := ConnectionCaption + ' / ' + FDefaultSchema;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

type
  TCastThread = class(TThread);
  
procedure TMySQLConn.FetchData(KindOfData: Integer; FetchMethod: TThreadExecMethod; RefreshMethod: TThreadExecMethod;
  Target: TObject; StatusBarText: WideString; AllowParallelExecution: Boolean; AllowDuplicatedKindOfData: Boolean;
  ControlThread: TThread);

// This method starts retrival of data by creating a separate thread and using the provided fetch and refresh methods.
// KindOfData determines what must be retrieved and is used in conjunction with AllowDuplicateKindOfData to allow or
// deny certain data retrieval actions.
// ControlThread has a special meaning as it is used to cancel long lasting operations. If it is assigned this method
// waits for the background to return unless the control thread gets terminated. In this case the data thread
// is killed and this method immediately returns to the caller.

var
  WaitResult: Cardinal;
  Thread: TFetchDataThread;
  
begin
  if (not (KindOfData in DataBeingFetched)) or
    (AllowDuplicatedKindOfData) then
  begin
    //Add KindOfData to DataBeingFetched Set
    if (not (KindOfData in DataBeingFetched)) then
      DataBeingFetched := DataBeingFetched + [KindOfData];
    try
      // Create a new thread.
      Thread := TFetchDataThread.Create(self, KindOfData, FetchMethod, RefreshMethod, Target, StatusBarText,
        AllowParallelExecution);
      FWorkList.Add(Thread);
      try
        // Thread will be freed after execution if no control thread is given.
        if ControlThread = nil then
          Thread.FreeOnTerminate := True;

        Thread.Resume;

        if Assigned(ControlThread) then
        begin
          // Wait for the data thread to finish but don't deadlock. Instead poll the thread until it is finished
          // or the control thread is terminated.
          WaitResult := WAIT_OBJECT_0;
          while not TCastThread(ControlThread).Terminated do
          begin
            WaitResult := WaitForSingleObject(Thread.Handle, 100);
            if WaitResult = WAIT_OBJECT_0 then
              Break;
          end;
          // Kill the fetch thread if the control thread has been told to finish but the fetch thread is still working.
          if WaitResult = WAIT_TIMEOUT then
          begin
            TerminateThread(Thread.Handle, 999);
            FetchDataThreadCriticalSection.Release;

            // Close the connection if we have to kill the last query.
            Reconnect;
           end;
          Thread.Free;
        end;
      except
        FWorkList.Remove(Thread);
        Thread.Free;

        raise;
      end;
    except
      DataBeingFetched := DataBeingFetched - [KindOfData];

      raise;
    end;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.StopCurrentThread(Timeout: Integer);

begin
  if Assigned(FCurrentDataThread) then
  begin
    FCurrentDataThread.Terminate;
    if WaitForSingleObject(FCurrentDataThread.Handle, Timeout) = WAIT_TIMEOUT then
      TerminateThread(FCurrentDataThread.Handle, 999);
    DataBeingFetched := DataBeingFetched - [FCurrentDataThread.KindOfData];
    FreeAndNil(FCurrentDataThread);
    FetchDataThreadCriticalSection.Release;
    if Assigned(FStatusBar) then
    begin
      FStatusBar.Panels[1].Text := '';
      FStatusBar.Invalidate;
    end;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.Reconnect: Boolean;

var
  ConnectionResult: Integer;

begin
  Result := False;

  if (User_Connection = nil) then
    Exit;

  if (FMySQL <> nil) then
    Disconnect;

  FMySQL := myx_mysql_init();
  if (FMySQL = nil) then
    raise EMyxError.Create('Error while allocating memory for MySQL Struct.');

  ConnectionResult := myx_connect_to_instance(
    User_Connection.get_record_pointer,
    FMySQL);
  if (ConnectionResult <> 0) then
    Exit;

  Connected := True;
  MessageToAllForms(WM_Reconnected, 0, 0);

  Result := True;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.Disconnect;

begin
  //Close connection and free memory for MySQL struct
  myx_mysql_close(FMySQL);
  FMySQL := nil;

  Connected := False;
  MessageToAllForms(WM_Disconnected, 0, 0);
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.CheckConnectionStatusChange(SQLCmd: WideString);

  //----------------------------------------------------------------------------

  function IsEqual(Keyword: PChar; S: PWideChar): Boolean;

  // Determines if the given string S is (case insensitively) the same as the
  // given keyword. The string might end with space chars.

  begin
    // We are comparing ASCII strings here so we can safely cast to char
    // for quick case conversion.
    while (Keyword^ <> #0) and (Upcase(Char(S^)) = Keyword^) do
    begin
      Inc(Keyword);
      Inc(S);
    end;
    Result := (Keyword^ = #0) and (S^ in [WideChar(#0), WideChar(' ')]);
  end;

  //----------------------------------------------------------------------------

var
  Head: PWideChar;
  FoundTransactionCommand: Boolean;

begin
  // Catch transaction status change.
  Head := PWideChar(SQLCmd);
  while Head^ = ' ' do
    Inc(Head);

  // Examine first non-space letter for a quick decision.
  FoundTransactionCommand := False;
  case Head^ of
    'S', 's':
      if IsEqual('START TRANSACTION', Head) then
      begin
        InTransaction := True;
        FoundTransactionCommand := True;
      end;
    'C', 'c':
      if IsEqual('COMMIT', Head) then
      begin
        InTransaction := False;
        FoundTransactionCommand := True;
      end;
    'R', 'r':
      if IsEqual('ROLLBACK', Head) then
      begin
        InTransaction := False;
        FoundTransactionCommand := True;
      end;
  end;

  // When in transaction, check if the executed command auto-commits the Trx
  if not FoundTransactionCommand and InTransaction then
    InTransaction := (myx_check_whether_commits_transaction(FMySQL, SQLCmd) = 0);
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.ExecuteDirect(SQLCmd: WideString; DisplayErrorDialog: Boolean): Boolean;

var
  ErrorCode: MYX_LIB_ERROR;
  MySQLErrorNr: Integer;

begin
  FetchDataThreadCriticalSection.Acquire;
  try
    ExecuteDirect := False;
    try
      myx_query_execute_direct(FMySQL, SQLCmd,
        @ErrorCode);

      if (ErrorCode = MYX_NO_ERROR) then
      begin
        ExecuteDirect := True;

        //Check if command causes a transaction state change
        CheckConnectionStatusChange(SQLCmd);
      end
      else
      begin
        MySQLErrorNr := myx_mysql_errno(FMySQL);

        if (MySQLErrorNr = CR_SERVER_GONE_ERROR) or
          (MySQLErrorNr = CR_SERVER_LOST) or
          (MySQLErrorNr = CR_CONN_HOST_ERROR) then
        begin
          Disconnect;
        end;

        if (DisplayErrorDialog) then
          ShowModalDialog('Execution Error',
            'Error while executing query.' + #13#10#13#10 +
            SQLCmd + #13#10#13#10 +
            'MySQL Error Number ' + IntToStr(MySQLErrorNr) + #13#10 +
            myx_mysql_error(FMySQL),
            myx_mtError, 'OK');
      end;
    except
      on x: Exception do
        if (DisplayErrorDialog) then
          ShowModalDialog('Execution Error',
            'The following error occured while executing the query.' + #13#10#13#10 +
            SQLCmd + #13#10#13#10 +
            x.Message, myx_mtError, 'OK');
    end;
  finally
    FetchDataThreadCriticalSection.Release;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.SetDefaultSchema(DefaultSchema: WideString);

var
  error: Integer;

begin
  if FMySQL <> nil then
  begin
    if DefaultSchema <> '' then
    begin
      error := myx_use_schema(FMySQL, DefaultSchema);
      if (error = 0) then
        FDefaultSchema := myx_get_default_schema(FMySQL)
      else
        raise EMyxSQLError.Create(Format(
          _('The default schema cannot be changed to %s'), [DefaultSchema]),
          myx_mysql_errno(FMySQL), myx_mysql_error(FMySQL));
    end
    else
      FDefaultSchema := '';

    User_Connection.schema := DefaultSchema;

    RefreshConnectionCaption;

    MessageToAllForms(WM_DefaultSchemaChanged, 0, 0);
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.SchemaListChanged;

begin
  MessageToAllForms(WM_SchemaListChanged, 0, 0);
end;

//----------------------------------------------------------------------------------------------------------------------

procedure MySQLQueryPostHook(mysql: Pointer; cdata: Pointer; query: PChar; length: Integer); cdecl;

var
  PSender: TMySQLConn;

begin
  PSender := cdata;

  PSender.QueryHookMultiReadSync.BeginWrite;
  try
    PSender.QueryHookSQL := UTF8Decode(query);
  finally
    PSender.QueryHookMultiReadSync.EndWrite;
  end;

  TThread.Synchronize(nil, PSender.TransactionQueryExecuted);
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.SetInTransaction(InTransaction: Boolean);

var
  I: Integer;
  TransactionStarted: Boolean;

begin
  if FInTransaction <> InTransaction then
  begin
    TransactionStarted := ((Not (FInTransaction)) and (InTransaction));

    FInTransaction := InTransaction;

    if Assigned(FTransactionStatusChangeListeners) then
      for I := 0 to FTransactionStatusChangeListeners.Count - 1 do
      begin
        if (TransactionStarted) then
          IMySQLConnTransactionStatusChangeListener(FTransactionStatusChangeListeners[I]).TransactionStarted;
          
        IMySQLConnTransactionStatusChangeListener(FTransactionStatusChangeListeners[I]).TransactionStatusChanged;
      end;

    if FInTransaction then
      myx_mysql_set_query_hooks(FMySQL, nil, @MySQLQueryPostHook, Self)
    else
      myx_mysql_set_query_hooks(FMySQL, nil, nil, nil);
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.GetQuoteChar: WideString;

begin
  if (FQuoteChar='') then
    FQuoteChar := Chr(myx_get_mysql_quote_char(FMySQL, 0));

  Result := FQuoteChar;
end;

//----------------------------------------------------------------------------------------------------------------------

function TMySQLConn.GetFetchingData: Boolean;

begin
  Result := (FCurrentDataThread<>nil);
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.AddTransactionStatusListener(Listener: IMySQLConnTransactionStatusChangeListener);

begin
  if FTransactionStatusChangeListeners = nil then
    FTransactionStatusChangeListeners := TList.Create;
  if FTransactionStatusChangeListeners.IndexOf(Pointer(Listener)) = -1 then
    FTransactionStatusChangeListeners.Add(Pointer(Listener));
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.RemoveTransactionStatusListener(Listener: IMySQLConnTransactionStatusChangeListener);

begin
  if FTransactionStatusChangeListeners <> nil then
  begin
    FTransactionStatusChangeListeners.Remove(Pointer(Listener));

    if FTransactionStatusChangeListeners.Count = 0 then
    begin
      FTransactionStatusChangeListeners.Free;
      FTransactionStatusChangeListeners := nil;
    end;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TMySQLConn.TransactionQueryExecuted;

var
  I: Integer;

begin
  QueryHookMultiReadSync.BeginRead;
  try
    if Assigned(FTransactionStatusChangeListeners) then
      for I := 0 to FTransactionStatusChangeListeners.Count - 1 do
        IMySQLConnTransactionStatusChangeListener(
          FTransactionStatusChangeListeners[I]).TransactionQueryExecuted(QueryHookSQL);
  finally
    QueryHookMultiReadSync.EndRead;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

constructor TFetchDataThread.Create(AMySQLConn: TMySQLConn; AKindOfData: Integer; AFetchMethod: TThreadExecMethod;
  ARefreshMethod: TThreadExecMethod; ATarget: TObject; AStatusBarText: WideString; AAllowParallelExecution: Boolean);

begin
  inherited Create(True);

  FConnection := AMySQLConn;
  self.KindOfData := AKindOfData;
  self.FetchMethod := AFetchMethod;
  self.RefreshMethod := ARefreshMethod;
  FTarget := ATarget;
  self.StatusBarText := AStatusBarText;
  SynchronizedMethod := nil;
  AllowParallelExecution := AAllowParallelExecution;
end;

//----------------------------------------------------------------------------------------------------------------------

destructor TFetchDataThread.Destroy;

begin
  StatusBarText := '';
  Synchronize(UpdateStatusBar);

  inherited Destroy;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TFetchDataThread.Execute;

begin
  if (not (AllowParallelExecution)) then
    FConnection.FetchDataThreadCriticalSection.Acquire;
  try
    try
      with FConnection.FWorkList.LockList do
      try
        Remove(Self);
      finally
        FConnection.FWorkList.UnlockList;
      end;

      if not Terminated then
      begin
        FConnection.FCurrentDataThread := Self;

        if (FStatusBarText <> '') then
          Synchronize(UpdateStatusBar);

        if (Assigned(FetchMethod)) then
          FetchMethod(self);

        if (Assigned(RefreshMethod)) then
          ExecuteSynchronized(RefreshMethod);
      end;
    except
      on x: EMyxSQLError do
      begin
        if (x.ErrorNr = CR_SERVER_GONE_ERROR) or
          (x.ErrorNr = CR_SERVER_LOST) or
          (x.ErrorNr = CR_CONN_HOST_ERROR) then
        begin
          FConnection.Disconnect;
        end;

        ErrorMsg := x.FormattedMessage;
        Synchronize(ShowError);
      end;
      on x: Exception do
      begin
        ErrorMsg := x.Message;
        Synchronize(ShowError);
      end;
    end;
  finally
    FConnection.FCurrentDataThread := nil;
    FConnection.DataBeingFetched := FConnection.DataBeingFetched - [KindOfData];
    if (not (AllowParallelExecution)) then
      FConnection.FetchDataThreadCriticalSection.Release;

    Synchronize(UpdateStatusBar);
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TFetchDataThread.ReleaseCriticalSection;

begin
  FConnection.FetchDataThreadCriticalSection.Release;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TFetchDataThread.UpdateStatusBar;

begin
  if Assigned(FConnection.FStatusBar) then
  begin
    if FConnection.FStatusBar.Panels.Count > 1 then
    begin
      FConnection.FStatusBar.Panels[1].Text := FStatusBarText;
      FConnection.FStatusBar.Repaint;
    end;
  end;
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TFetchDataThread.ShowError;

begin
  ShowModalDialog(Application.Title + ' ' + _('Error'), ErrorMsg,
    myx_mtError, _('OK'));
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TFetchDataThread.ExecuteSynchronized(SynchronizedMethod: TThreadExecMethod);

begin
  self.SynchronizedMethod := SynchronizedMethod;

  Synchronize(ExecuteSynchronizedMethod);
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TFetchDataThread.ExecuteSynchronizedMethod;

begin
  if Assigned(SynchronizedMethod) then
    SynchronizedMethod(self);
end;

//----------------------------------------------------------------------------------------------------------------------

procedure TFetchDataThread.SetStatusBar(StatusBarText: WideString);

begin
  FStatusBarText := StatusBarText;

  UpdateStatusBar;
end;

//----------------------------------------------------------------------------------------------------------------------

end.