with Library;
with Calendar;
use Calendar;
with Daemon;
with Direct_Io;
with List_Generic;
with Time_Utilities;
with String_Utilities;
with Directory_Tools;
with Io;
with Log;
with Profile;
package body Batch_Stream is

    type Job_Id_Array is array (Job_Identifier) of Boolean;

    Stream_Directory : constant String := "!Machine.Batch_Streams.";

    package Stream_Io is new Direct_Io (Job_Descriptor);

    In_Use : Job_Id_Array := (others => False);

    function Make (Job_Number : Job_Identifier;
                   Command : String;
                   Context : String;
                   User_Name : String;
                   Error_Filename : String;
                   Input_Filename : String;
                   Output_Filename : String;
                   Start_Time : Time_Utilities.Time;
                   Restartable : Boolean;
                   Response : String := "<PROFILE>") return Job_Descriptor is
    begin
        return Job_Descriptor'
                  (Number => Job_Number,
                   Command => Bounded_String.Value (Command, Job_String_Length),
                   Context => Bounded_String.Value (Context, Job_String_Length),
                   Username => Bounded_String.Value
                                  (User_Name, Job_String_Length),
                   Error_Filename =>
                      Bounded_String.Value (Error_Filename, Job_String_Length),
                   Input_Filename =>
                      Bounded_String.Value (Input_Filename, Job_String_Length),
                   Output_Filename =>
                      Bounded_String.Value (Output_Filename, Job_String_Length),
                   Response => Bounded_String.Value
                                  (Response, Job_String_Length),
                   Start_Time => Start_Time,
                   State => Waiting,
                   Restartable => Restartable);
    end Make;

    procedure Set_State (Of_Job : in out Job_Descriptor; To : Job_State) is
    begin
        Of_Job.State := To;
    end Set_State;

    function Number (Of_Job : Job_Descriptor) return Job_Identifier is
    begin
        return Of_Job.Number;
    end Number;

    function Command (Of_Job : Job_Descriptor) return String is
    begin
        return Bounded_String.Image (Of_Job.Command);
    end Command;

    function Context (Of_Job : Job_Descriptor) return String is
    begin
        return Bounded_String.Image (Of_Job.Context);
    end Context;

    function Username (Of_Job : Job_Descriptor) return String is
    begin
        return Bounded_String.Image (Of_Job.Username);
    end Username;

    function Error_Filename (Of_Job : Job_Descriptor) return String is
    begin
        return Bounded_String.Image (Of_Job.Error_Filename);
    end Error_Filename;

    function Input_Filename (Of_Job : Job_Descriptor) return String is
    begin
        return Bounded_String.Image (Of_Job.Input_Filename);
    end Input_Filename;

    function Output_Filename (Of_Job : Job_Descriptor) return String is
    begin
        return Bounded_String.Image (Of_Job.Output_Filename);
    end Output_Filename;

    function Response (Of_Job : Job_Descriptor) return String is
    begin
        return Bounded_String.Image (Of_Job.Response);
    end Response;

    function Start_Time (Of_Job : Job_Descriptor) return Time_Utilities.Time is
    begin
        return Of_Job.Start_Time;
    end Start_Time;

    function Restartable (Of_Job : Job_Descriptor) return Boolean is
    begin
        return Of_Job.Restartable;
    end Restartable;

    function State (Of_Job : Job_Descriptor) return Job_State is
    begin
        return Of_Job.State;
    end State;

    procedure Set_Number (Of_Job : in out Job_Descriptor;
                          To : Job_Identifier) is
    begin
        Of_Job.Number := To;
    end Set_Number;


    function Assign_Job_Id return Job_Identifier is
        Found_One : Boolean := False;
        Result : Job_Identifier;
    begin
        for I in reverse Job_Identifier loop
            Result := I;
            Found_One := not In_Use (I);
            exit when Found_One;
        end loop;  
        if Found_One then
            In_Use (Result) := True;
            return Result;
        else
            raise Queue_Is_Full;
        end if;
    end Assign_Job_Id;

    procedure Free (Job_Id : Job_Identifier) is
    begin
        In_Use (Job_Id) := False;
    end Free;

    function Get_Name (Of_Stream_Number : Natural) return String is
    begin
        return Stream_Directory & "stream" &
                  String_Utilities.Strip_Leading
                     (Natural'Image (Of_Stream_Number));
    end Get_Name;


    procedure Take_Snapshot (Stream_Number : Natural) is
        The_File : Io.File_Type;
    begin
        --Io.Open (File => The_File,
        --         Mode => Io.Out_File,
        --         Name => Get_Name (Stream_Number),
        --         Form => "");
        --Io.Save (File => The_File);
        --Io.Close (File => The_File);
        Daemon.Schedule (Client => "SNAPSHOT",
                         Interval => Daemon.Interval ("SNAPSHOT"),
                         First_Run => 0.0,
                         Response => "<PROFILE>");
    end Take_Snapshot;
    procedure Initialize_Stream (Stream_Number : Natural) is
        The_Stream : Stream_Io.File_Type;
    begin
        declare
            use Directory_Tools;
            Hndl : Object.Handle;
            Status : Object.Error_Code;
        begin
            Hndl := Naming.Resolution (Naming.Prefix
                                          (Stream_Directory & "foo"));
            if Object.Is_Bad (Object.Err_Code (Hndl)) then
                if Object."=" (Object.Category (Object.Err_Code (Hndl)),
                               Object.Undefined_Name) then
                    Library.Create (Name =>
                                       Naming.Prefix (Stream_Directory & "foo"),
                                    Kind => Library.Directory,
                                    Vol => Library.Nil,
                                    Model => "!Model.R1000",
                                    Response => "<PROFILE>");
                    --Library_Object.Create
                    --   (Name => Naming.Prefix (Stream_Directory & "foo"),
                    --    Kind => Library_Object.Directory,
                    --    New_Library => Hndl,
                    --    Status => Status);
                    --if Object.Is_Bad (Status) then
                    --    Log.Put_Line (Object.Message (Status),
                    --                  Profile.Error_Msg);
                    --    Log.Put_Line ("Create streams directory failed",
                    --                  Profile.Negative_Msg);
                    --    raise File_Access_Error;
                    --end if;
                else
                    Log.Put_Line
                       ("Unknown error creating batch_streams directory",
                        Profile.Error_Msg);
                    raise File_Access_Error;
                end if;
            end if;
        end;
        Stream_Io.Open (File => The_Stream,
                        Mode => Stream_Io.Inout_File,
                        Name => Get_Name (Stream_Number),
                        Form => "");  
        Stream_Io.Close (The_Stream);
    exception
        when Stream_Io.Name_Error =>
            Stream_Io.Create (File => The_Stream,
                              Mode => Stream_Io.Inout_File,
                              Name => Get_Name (Stream_Number),
                              Form => "");
            Stream_Io.Close (The_Stream);
    end Initialize_Stream;

    procedure Delete_Stream (Stream_Number : Natural) is
        The_Stream : Stream_Io.File_Type;

    begin
        Stream_Io.Open (File => The_Stream,
                        Mode => Stream_Io.Inout_File,
                        Name => Get_Name (Stream_Number),
                        Form => "");  
        Stream_Io.Delete (The_Stream);
    exception
        when Stream_Io.Name_Error =>
            null;
    end Delete_Stream;

    procedure Open_Or_Wait (For_File : in out Stream_Io.File_Type;
                            Mode : Stream_Io.File_Mode;
                            Name : String;
                            Form : String) is
        Max_Retry_Count : constant Natural := 10;
        Retry_Count : Natural := 0;
        Opened : Boolean := False;
    begin
        loop
            begin
                Stream_Io.Open
                   (File => For_File, Mode => Mode, Name => Name, Form => "");
                Opened := True;
            exception
                when Stream_Io.Status_Error =>
                    delay 5.0;
                    Retry_Count := Retry_Count + 1;
                when Stream_Io.Name_Error =>
                    Retry_Count := Max_Retry_Count + 1;
            end;
            exit when Opened or (Retry_Count > Max_Retry_Count);
        end loop;
        if not Opened then
            raise File_Access_Error;
        end if;
    end Open_Or_Wait;

    procedure Add_To_Stream (Stream_Number : Natural;
                             The_Job : in out Job_Descriptor) is
        use Stream_Io;
        Found_One : Boolean;
        The_Stream : Stream_Io.File_Type;
        Read_Job : Job_Descriptor;
        Read_Job_State : Job_State;
        Was_Open : Boolean := False;
    begin

        Set_Number (Of_Job => The_Job,  
                    To => Assign_Job_Id);
        Open_Or_Wait (For_File => The_Stream,
                      Mode => Stream_Io.Inout_File,
                      Name => Get_Name (Stream_Number),
                      Form => "");
        while not Stream_Io.End_Of_File (The_Stream) loop
            Stream_Io.Read (The_Stream, Read_Job);
            Read_Job_State := State (Read_Job);
            Found_One := Read_Job_State = Cancelled or Read_Job_State = Done;
            if Found_One then  
                Stream_Io.Set_Index (File => The_Stream,
                                     To => Stream_Io.Index (The_Stream) - 1);
                exit;
            end if;
        end loop;  
        Stream_Io.Write (File => The_Stream, Item => The_Job);
        Stream_Io.Close (The_Stream);
        Take_Snapshot (Stream_Number);
    end Add_To_Stream;

    procedure Change_State (Of_Job_Number : Job_Identifier;
                            To : Job_State;  
                            In_Stream : Natural;
                            Successful : out Boolean) is

        use Stream_Io;
        The_Stream : Stream_Io.File_Type;
        Found_One : Boolean;
        Read_Job : Job_Descriptor;
    begin
        Open_Or_Wait (For_File => The_Stream,
                      Mode => Stream_Io.Inout_File,
                      Name => Get_Name (In_Stream),
                      Form => "");  
        while not Stream_Io.End_Of_File (The_Stream) loop
            Stream_Io.Read (The_Stream, Read_Job);
            Found_One := Number (Read_Job) = Of_Job_Number;
            if Found_One then  
                case To is
                    when Cancelled | Done =>
                        Free (Number (Read_Job));
                    when Waiting | Running =>
                        null;
                end case;
                Stream_Io.Set_Index (File => The_Stream,
                                     To => Stream_Io.Index (The_Stream) - 1);
                Set_State (Read_Job, To);
                Stream_Io.Write (File => The_Stream, Item => Read_Job);
                exit;
            end if;
        end loop;  
        Stream_Io.Close (The_Stream);
        if Found_One then
            Take_Snapshot (In_Stream);  
        end if;
        Successful := Found_One;
    end Change_State;

    procedure Get_Next_Job_To_Run (From_Stream : Natural;
                                   The_Job : out Job_Descriptor;
                                   Found_One : out Boolean) is
        The_Stream : Stream_Io.File_Type;
        Read_Job : Job_Descriptor;
        Next_Job : Job_Descriptor := Nil;
        Got_One : Boolean := False;
    begin
        Open_Or_Wait (For_File => The_Stream,
                      Mode => Stream_Io.In_File,
                      Name => Get_Name (From_Stream),
                      Form => "");  
        while not Stream_Io.End_Of_File (The_Stream) loop
            Stream_Io.Read (The_Stream, Read_Job);
            if State (Read_Job) = Waiting then
                if not Got_One then
                    Next_Job := Read_Job;
                    Got_One := True;  
                else
                    if Time_Utilities.Convert_Time (Start_Time (Read_Job)) <
                       Time_Utilities.Convert_Time (Start_Time (Next_Job)) then
                        Next_Job := Read_Job;
                    end if;
                end if;
            end if;
        end loop;  
        Stream_Io.Close (The_Stream);
        The_Job := Next_Job;
        Found_One := Got_One;
    end Get_Next_Job_To_Run;

    procedure Initialize (Stream_Number : Natural;
                          The_Iter : in out Iterator) is
        The_Stream : Stream_Io.File_Type;
        Result_Iter : Iterator := Iterator (Batch_List.Nil);
        Read_Job : Job_Descriptor;
    begin
        Open_Or_Wait (For_File => The_Stream,
                      Mode => Stream_Io.In_File,
                      Name => Get_Name (Stream_Number),
                      Form => "");  
        while not Stream_Io.End_Of_File (The_Stream) loop
            Stream_Io.Read (The_Stream, Read_Job);  
            Result_Iter := Make (Read_Job, Result_Iter);
        end loop;  
        Stream_Io.Close (The_Stream);
        The_Iter := Result_Iter;
    end Initialize;

    function Value (Of_Iter : Iterator) return Job_Descriptor is
    begin
        return First (Of_Iter);
    end Value;

    procedure Next (In_Iter : in out Iterator) is
    begin
        In_Iter := Rest (In_Iter);
    end Next;

    function Done (For_Iter : Iterator) return Boolean is
    begin
        return Is_Empty (For_Iter);
    end Done;
    procedure Traverse_Streams is
        use Directory_Tools;
        Iter : Object.Iterator;
        Stream_Number : Natural;
        Ok : Boolean;
    begin
        Iter := Naming.Resolution (Stream_Directory & ("stream@"));
        while not Object.Done (Iter) loop
            declare
                Stream_Simple_Name : constant String :=
                   Naming.Simple_Name (Object.Value (Iter));
            begin
                String_Utilities.String_To_Number
                   (Source => Stream_Simple_Name
                                 (Stream_Simple_Name'First + 6 ..
                                     Stream_Simple_Name'Last),
                    Target => Stream_Number,
                    Worked => Ok,
                    Base => 10);
                if Ok then
                    Visit (Stream_Number);
                end if;
            end;
            Object.Next (Iter);
        end loop;
    end Traverse_Streams;
end Batch_Stream;