Page History

using System;
using System.Collections.Immutable;
using Akka;
using Akka.Actor;
using Akka.Event;
using Akka.Monitoring;
using Akka.Streams;
using Akka.Streams.Dsl;
using AkkaNetCore.Models.Message;

namespace AkkaNetCore.Actors.Utils
{
    // 일괄 처리(데이터 인입)    
    public class ThrottleWork : ReceiveActor
    {
        private readonly ILoggingAdapter logger = Context.GetLogger();

        private IActorRef consumer;

        public ThrottleWork(int element,int maxBust)
        {
            
            ReceiveAsync<SetTarget>(async target =>
            {
                consumer = target.Ref;
            });

            ReceiveAsync<object>(async message =>
            {
    using System;
using System.Collections.Immutable;
using Akka;
using Akka.Actor;
using Akka.Event;
using Akka.Monitoring;
using Akka.Streams;
using Akka.Streams.Dsl;
using AkkaNetCore.Models.Message;

namespace AkkaNetCore.Actors.Utils
{
    // 일괄 처리(데이터 인입)    
    public class ThrottleWork : ReceiveActor
    {
        private readonly ILoggingAdapter logger =if Context.GetLogger();

  (message is Batch batchMessage)
      private IActorRef consumer;

        public{
 ThrottleWork(int element,int maxBust)
        {
         int Count = batchMessage.Obj.Count;
            ReceiveAsync<SetTarget>(async target =>
      Context.IncrementMessagesReceived();
      {
              Source<object, NotUsed> consumersource = target.RefSource.From(batchMessage.Obj);

            });

           using ReceiveAsync<object>(asyncvar messagematerializer =>
 Context.Materializer())
                    {
                if  (message is Batch batchMessage)
   var factorials = source;
          {
              factorials
      int Count = batchMessage.Obj.Count;
                    Context.IncrementMessagesReceived();
                //.ZipWith(Source.From(Enumerable.Range(0, 100)), (num, idx) => $"{idx}! = {num}")
    Source<object, NotUsed> source = Source.From(batchMessage.Obj);

                    using .Throttle(var materializer = Context.Materializer())element, TimeSpan.FromSeconds(1), maxBust, ThrottleMode.Shaping)
                    {
         .RunForeach(obj => {             var factorials = source;
                 
       factorials
                          var nowstr = //.ZipWith(Source.From(Enumerable.Range(0, 100)), (num, idx) => $"{idx}! = {num}")DateTime.Now.ToString("mm:ss");
                             .Throttle(element, TimeSpan.FromSeconds(1), maxBust, ThrottleMode.Shaping)
    if(obj is DelayMsg delayMsg)
                               .RunForeach(obj => {                                     
                                  var nowstr = DateTimeConsole.Now.ToStringWriteLine("mm:ss$"[{nowstr}] -  {delayMsg.Message}");
                                     if (objconsumer is!= DelayMsg null) consumer.Tell(delayMsg);
                                 {}                                     
                                     Console.WriteLine($"[{nowstr}] -  {delayMsg.Message}");
           }, materializer)
                         if (consumer != null) consumer.TellWait(delayMsg);
                    }
                }
            });
        }
    }


    public class ThrottleActor : FSM<State, IData>
    {
        private readonly ILoggingAdapter logger = Context.GetLogger();
        private int CollectSec;

        public ThrottleActor(int _CollectSec)
       }, materializer){
            CollectSec = _CollectSec;

               .Wait(StartWith(State.Idle, Uninitialized.Instance);

            When(State.Idle, state   =>
    }
        {
        }
            });
        }
    }


if (state.FsmEvent is SetTarget target && state.StateData is Uninitialized)
       public class ThrottleActor : FSM<State, IData>
    {
        private readonly ILoggingAdapter logger = Context.GetLogger();
        private int CollectSec;

  return Stay().Using(new Todo(target.Ref, ImmutableList<object>.Empty));
      public ThrottleActor(int _CollectSec)
        {}

             CollectSec   =return _CollectSecnull;

            StartWith(State.Idle, Uninitialized.Instance});

            When(State.IdleActive, state =>
            {
                if ((state.FsmEvent is Flush || state.FsmEvent is SetTarget target StateTimeout)
                    && state.StateData is isTodo Uninitializedt)
                {
                    return StayGoTo(State.Idle).Using(new Todo(target.Ref, t.Copy(ImmutableList<object>.Empty));
                }

                return null;
            }, TimeSpan.FromSeconds(CollectSec));

            WhenWhenUnhandled(State.Active, state =>
            {
                if ((state.FsmEvent is Flush || state.FsmEvent is StateTimeout)
                    FsmEvent is Queue q && state.StateData is Todo t)
                {
                    return GoTo(State.IdleActive).Using(t.Copy(ImmutableList<object>.Emptyt.Queue.Add(q.Obj)));
                }
                else
                {
              return null;
     logger.Warning($"Received unhandled request {state.FsmEvent} in state  }, TimeSpan.FromSeconds(CollectSec));

{StateName}/{state.StateData}");
            WhenUnhandled(state =>
       return  Stay();
   {
             }
   if (state.FsmEvent is Queue q && state.StateData is Todo t});

            OnTransition((initialState,    {nextState) =>
            {
        return GoTo(State.Active).Using(t.Copy(t.Queue.Add(q.Obj)));
       if (initialState == State.Active && nextState    }== State.Idle)
                else{
                {
    if (StateData is Todo todo)
             logger.Warning($"Received unhandled request {state.FsmEvent} in state {StateName}/{state.StateData}");
       {
                      return Stay(  todo.Target.Tell(new Batch(todo.Queue));
                    }
              });

      else
      OnTransition((initialState, nextState) =>
            {
                    if (initialState == State.Active && nextState == State.Idle) // nothing to do
                {
    }
                if}
 (StateData is Todo todo)
        });

            {Initialize();
        }
                todo.Target.Tell(new Batch(todo.Queue));
                    }
                    else
                    {
}
}

// FSM이 사용되었으며, 특정초마다 수집한 데이터를 처리합니다. (스트림 출발점)
When(State.Active, state =>
{
.......	
}, TimeSpan.FromSeconds(CollectSec));

// 빗물담당자는 빗물을 한꺼번에 흘려보냈지만
// 작업자는 안정적으로 초당 5개씩 처리를 하며
// 흘러들어온 버퍼를 꾸준하게 작업한만큼 비웁니다.
Source<object, NotUsed> source = Source.From(batchMessage.Obj);
var factorials = source;
factorials	 
	 .Throttle(element, TimeSpan.FromSeconds(1), maxBust, ThrottleMode.Shaping)
	 .RunForeach(obj => {
		 var nowstr = DateTime.Now.ToString("mm:ss");
		 if(obj is DelayMsg delayMsg)
		 {                        // nothing to do
                    }
       
			 Console.WriteLine($"[{nowstr}] -  {delayMsg.Message}");
			 if (consumer != null) consumer.Tell(delayMsg);
		 }         }
            });

            Initialize();
	        }
    }
}}, materializer)
	 .Wait();