feat: support appending batch messages using hstreamdb_producer:appen…

…d/2 (#74)

* feat: support appending batch messages using hstreamdb_producer:append/2

* fix: fix the appup file

src/hstreamdb.erl

@@ -116,8 +116,8 @@ append(Producer, PartitioningKey, PayloadType, Payload) ->
     Record = to_record(PartitioningKey, PayloadType, Payload),
     append(Producer, Record).
 
-append(Producer, Record) ->
-    hstreamdb_producer:append(Producer, Record).
+append(Producer, Data) ->
+    hstreamdb_producer:append(Producer, Data).
 
 flush(Producer) ->
     hstreamdb_producer:flush(Producer).

src/hstreamdb_erl.appup.src

@@ -1,15 +1,13 @@
 %% -*- mode: erlang -*-
 {VSN,
-  [{"0.2.4",
+  [{"0.3.1+v0.13.0",
     [{load_module,hstreamdb,brutal_purge,soft_purge,[]},
-     {load_module,hstreamdb_channel_mgr,brutal_purge,soft_purge,[]},
      {load_module,hstreamdb_producer,brutal_purge,soft_purge,[]}
     ]},
-   {<<".*">>,[]}],
-  [{"0.2.4",
+   {<<".*">>,[{restart_application,hstreamdb_erl}]}],
+  [{"0.3.1+v0.13.0",
     [{load_module,hstreamdb,brutal_purge,soft_purge,[]},
-     {load_module,hstreamdb_channel_mgr,brutal_purge,soft_purge,[]},
      {load_module,hstreamdb_producer,brutal_purge,soft_purge,[]}
     ]},
-   {<<".*">>,[]}]
+   {<<".*">>,[{restart_application,hstreamdb_erl}]}]
 }.

src/hstreamdb_producer.erl

@@ -81,11 +81,11 @@ stop(Producer) ->
     ok = ecpool:stop_sup_pool(Producer),
     ok = ecpool:stop_sup_pool(writer_name(Producer)).
 
-append(Producer, Record) ->
+append(Producer, Records) ->
     ecpool:with_client(
       Producer,
       fun(Pid) ->
-              gen_server:call(Pid, {append, Record})
+              gen_server:call(Pid, {append, Records})
       end).
 
 flush(Producer) ->
@@ -137,11 +137,17 @@ init(Options) ->
         current_batches = #{}
     }}.
 
-handle_call({append, Record}, _From, State) ->
+handle_call({append, Records}, _From, State) ->
     if_not_overflooded(
       fun() ->
-              NState = do_append(Record, State),
-              {reply, ok, NState}
+        case is_list(Records) of
+            false ->
+                {reply, ok, do_append(Records, State)};
+            true ->
+                {reply, ok, lists:foldl(fun(Record, AccState) ->
+                    do_append(Record, AccState)
+                  end, State, Records)}
+        end
       end,
       State);
 
@@ -213,7 +219,7 @@ do_append({PartitioningKey, Record},
                                   NRecords = [Record],
                                   NRecordMap = RecordMap#{ShardId => NRecords},
                                   NFlushDeadlineMap = FlushDeadlineMap#{
-                                                        ShardId => 
+                                                        ShardId =>
                                                         Interval + erlang:monotonic_time(millisecond)
                                                        },