anchor.diff:  
       session_setup(Worker2, SessId2, Cred2, Self),
   
  -    lists:foldl(fun(Evt, N) ->
  -        emit(Worker1, Evt#write_event{
  -            blocks = [#file_block{offset = N, size = 1}]
  -        }, SessId1),
  -        N + 1
  -    end, 0, lists:duplicate(6, #write_event{
  -        counter = 1, size = 1, file_size = 1
  -    })),
  -
  +    % Check whether subscription message has been sent to clients.
  +    ?assertMatch({ok, #write_event_subscription{}}, test_utils:receive_any(?TIMEOUT)),
  +    ?assertMatch({ok, #write_event_subscription{}}, test_utils:receive_any(?TIMEOUT)),
  +    ?assertEqual({error, timeout}, test_utils:receive_any()),
  +
  +    % Emit events.
  +    lists:foreach(fun(N) ->
  +        emit(Worker1, #write_event{size = 1, counter = 1, file_size = N + 1,
  +            blocks = [#file_block{offset = N, size = 1}]}, SessId1)
  +    end, lists:seq(0, 5)),
  +
  +    % Check whether handlers have been executed.
       ?assertMatch({ok, _}, test_utils:receive_msg({handler, [#write_event{
  -        counter = 6, size = 6, file_size = 1,
  +        counter = 6, size = 6, file_size = 6,
           blocks = [#file_block{offset = 0, size = 6}]
       }]}, ?TIMEOUT)),
  -    ?assertMatch({error, timeout}, test_utils:receive_any()),
  +    ?assertEqual({error, timeout}, test_utils:receive_any()),
   
       FileId1 = <<"file_id_1">>,
       FileId2 = <<"file_id_2">>,
   
  +    % Check aggregation of 'different' events.
       lists:foreach(fun(FileId) ->
           lists:foreach(fun(Evt) ->
               emit(Worker1, Evt, SessId2)
  
positive.code: end_per_suite(Config) ->
      test_node_starter:clean_environment(Config).
  
  %%%===================================================================
  %%% Internal functions
  %%%===================================================================
  
  %%--------------------------------------------------------------------
  %% @private
  %% @doc
  %% Creates new test session.
  %% @end
  %%--------------------------------------------------------------------
  -spec session_setup(Worker :: node(), SessId :: session:id(),
      Cred :: session:credentials(), Con :: pid()) -> ok.
  session_setup(Worker, SessId, Cred, Con) ->
      test_utils:mock_new(Worker, communicator),
      test_utils:mock_expect(Worker, communicator, send, fun
          (_, _) -> ok
      end),
      ?assertEqual({ok, created}, rpc:call(Worker, session_manager,
          reuse_or_create_session, [SessId, Cred, Con])).
  
  
negative.code: 
  -include("workers/datastore/models/session.hrl").
  -include("proto_internal/oneclient/common_messages.hrl").
  -include("proto_internal/oneclient/event_messages.hrl").
  -include_lib("ctool/include/logging.hrl").
  -include_lib("ctool/include/test/test_utils.hrl").
  -include_lib("ctool/include/test/assertions.hrl").
  
  %% export for ct
  -export([all/0, init_per_suite/1, end_per_suite/1, init_per_testcase/2,
      end_per_testcase/2]).
  
  %% tests
  -export([
      event_stream_test/1,
      event_manager_test/1
  ]).
  
  all() -> [
      event_stream_test,
      event_manager_test
  ].
  
  -define(TIMEOUT, timer:seconds(5)).
  