Streamkapis in the Business Intelligence Industry

Streamkap

streamkap

Who is this article for? The material is designed for developers with little experience with , and the main reason is to refresh their knowledge. Java Table of Contents: Variables and types It will be better to use these Switch and case. Try, catch, and finally. Lambdas, streams, optional? Record class Variables and types. Let’s start with the general and important things that you really should have learned earlier. Java, like many languages, has two types of variables: Primitives: Non-Primitive: Classes and arrays! Usually, we need to compare variables, but what methods should we use? “equals” or “== “ ? Let’s look at all the comparison methods available to us: ==      equal to
!=      not equal to
>       greater than
>=      greater than or equal to
<       less than
<=      less than or equal to This tricky question was asked in the interview. We can use it with classes, but: // Integer != int
int a = 1, b = 1;
System.out.println(a == b); // true

int c = 1, j = c;
System.out.println(c == j); // true

Integer ai = Integer.valueOf(1), bi = Integer.valueOf(1);
System.out.println(ai == bi); // true 

Integer ci = Integer.valueOf(128), ji = Integer.valueOf(128);
System.out.println(ci == ji); // false
/* You can always count on the fact that 
*  for values between -128 and 127, 
*  you get the identical Integer objects after autoboxing, 
*  and on some implementations you might get identical objects
*  even for higher values.
*/ It’s important to learn that the Integer class is not a primitive value because it is a regular class. Class is the root of the class hierarchy. Every class has as a superclass. All objects, including arrays, implement the methods of this class. Object Object The “==” operator compares the values of two object references to see if they refer to the same thing. For an object, we should use “equals”: // Indicates whether some other object is "equal to" this one.
public boolean equals(Object obj) Or interface for natural ordering (if you need to sort your collection with your own type of object): Comparable public class HugePoint implements Comparable<HugePoint> {

    // same as before

    @Override
    public int compareTo(HugePoint otherHugePoint) {
        return Integer.compare(getSize(), otherHugePoint.getSize());
    }

} The interface defines a method: Comparator compare(arg1, arg2) public class HugePointComparator implements Comparator<HugePoint> {

    @Override
    public int compare(HugePoint firstPoint, HugePoint secondPoint) {
       return Integer.compare(firstPoint.getSize(), secondPoint.getSize());
    }

} It’s easy to use Java 8+ this way: Comparator<Person> byAge = Comparator.comparing(Person::getAge);

Comparator byAge = (Person person1, Person person2) -> 
    Integer.compare(person1.getAge(), person2.getAge()); Which one do you need to use and when? I would recommend using “Comparable” interface for ordering (if you have a sorting method, etc.), and “Comparator” if you need to compare objects by a special field. It will be better to use these Date and time Why? For the current timestamp, just use . We don’t need to convert to milliseconds. Instant.now() // Instant class, one of the main classes of the Date-Time API, encapsulates a point on the timeline.
Instant.now();
Instant now = Instant.parse("2021-02-09T11:19:42.12Z");
Instant before = now.minus(Duration.ofDays(1)); For the local date format use the following: // LocalDateTime is an immutable date-time object 
// that represents a date-time, often viewed 
// as year-month-day-hour-minute-second. 
LocalDateTime.now();
LocalDateTime.of(2001, Month.FEBRUARY, 10, 03, 01); UUID // uuid
UUID uuid = UUID.randomUUID();
UUID uuid = new UUID(mostSignificant64Bits, leastSignificant64Bits); Random // random for int
int randomWithMathRandom = (int) ((Math.random() * (max - min)) + min);

// or easy way
Random random = new Random();
int randomWithNextInt = random.nextInt(); Switch and case. Try, catch, and finally When we have too many conditions, and they are all tied to the same type of variable in “if-then-else,” it is worth using a switch-case. // case 
int expression = 9;
switch(expression) {
  case 2:
    System.out.println("Small Size");
    break;

  case 3:
    System.out.println("Large Size");
    break;
        
  // default case
  default:
    System.out.println("Unknown Size");
} Also, up Java to version 19 or more, and we can use the expression: Day day = Day.WEDNESDAY;
var resultDay = switch (day) {
        case MONDAY, FRIDAY, SUNDAY -> 6;
        case TUESDAY                -> 7;
        case THURSDAY, SATURDAY     -> 8;
        case WEDNESDAY              -> 9;
        default -> throw new IllegalStateException("Invalid day: " + day);
    } 

System.out.println(resultDay); Don’t forget to use try/catch with methods that can throw exceptions. Try/catch and try with resources: // old
public int find(String playerFile) {
    try {
        Scanner contents = new Scanner(new File(playerFile));
        return Integer.parseInt(contents.nextLine());
    } catch (FileNotFoundException noFile) {
        throw new IllegalArgumentException("Couldn't find the file");
    }
}

// with resources
public int find(String playerFile) {
    try (Scanner contents = new Scanner(new File(playerFile))) {
      return Integer.parseInt(contents.nextLine());
    } catch (FileNotFoundException e ) {
      logger.warn("Couldn't find the file");
      return 0;
    }
}

// don't do it!
public int find(String playerFile) {
    try {
        // action
    } catch (Exception e) {} // <== catch and swallow
    return 0;
} Use , especially with channels: finally // close it!
public int find(String playerFile) throws FileNotFoundException {
    Scanner contents = null;

    try {
        contents = new Scanner(new File(playerFile));
        return Integer.parseInt(contents.nextLine());
    } finally {
        if (contents != null) {
            contents.close();
        }
    }
}

// Always close it!
public int find(String playerFile){
    MessageChannel channel = buildChannel(port);
    try {
      // do some action with channel
    } finally {
      channel.close(); // close!
    }
} Lambdas, streams, optional? The main reason for switching to Java 8 and higher is to be able to work with threads, streams, and functional interfaces. In this article, we will cover only working with streams, and we will deal with threads on the next page. First point — collections: Also, it would be good to go through the algorithms’ complexity table. I strongly recommend using this link: or the full article . gist here Let’s analyze the two approaches and see the difference. // old way: 
List<String> old = new ArrayList<>();
old.add("one");

// new way:
List<String> list = Arrays.asList("one", "two");
List<String> list = List.of("one", "two", "three");
Set<String> set = Set.of("one", "two", "three"); We can simplify working with the map. // old:
HashMap<Integer, String> hm = new HashMap<Integer, String>();
hm.put(1, "foo");

// new:
Map<String, String> map = Map.of("foo", "one", "bar", "two"); As you know, we have only two types of operations: and operations. intermediate terminal // Intermediate Operations
var stream = ...;
stream.filter();
stream.map();
stream.flatMap();
stream.distinct();
stream.sorted();
stream.peek();
stream.limit();
stream.skip();


// Terminal
stream.reduce();
stream.forEach();
stream.min();
stream.max();
stream.anyMatch();
stream.allMatch();
stream.noneMatch();
stream.findAny();
stream.findFirst();
stream.toArray();
stream.collect();
stream.count();
stream.forEachOrdered(); Just code: // Java 7 way
ArrayList results = new ArrayList(); 
for (Score score: scores) { 
  if (score.getVal() > 10) { 
      results.add(score.getVal()); 
   } 
} 
System.out.println(results);

// Java 8+ way
ArrayList results = scores.stream()
      .filter(score -> score.getVal() > 10)
      .collect(Collectors.toList()); 

System.out.println(results); If you have a duplicate code, just use functional interfaces: // Functional 
Consumer<Integer> consumer = (n) -> { System.out.println(n); };
numbers.forEach(consumer); Throwing exceptions every time an object is not found is not a good idea; you should use the “Optional” class for this: var empty = Optional.empty();
empty.isPresent(); // false

var notEmpty = Optional.of(name);
notEmpty.isPresent(); // true

String name = null;
Optional<String> opt = Optional.ofNullable(name);
opt.isPresent(); // false

String name = null;
String result = Optional.ofNullable(nullName).orElse("Default");

// get result
var name = Optional.of("Name");
name.get(); And if you really need to throw an exception, you should use “orElseThrow”: String name = Optional.ofNullable(name)
    .orElseThrow(IllegalArgumentException::new);

// throw NoSuchElementException.class
Optional.ofNullable(name).orElseThrow(); Record classes A typical situation: API had been written with “none thread safe” methods, and we didn’t know about the magic inside the library. What exactly do I mean? Any variables are mutable by default, which means we can change the value they hold. We don’t have read-only methods for classes inside; we can change them inside any threads: public class TestObject {
  private String string;
  // public contructor 
  // getters and setters
} In action: // create new object
ExecutorService executor = Executors.newFixedThreadPool(10);
var testObject = new UsefulObject();

executor.submit(() -> testObject.setTestString("test_1"));
executor.submit(() -> testObject.setTestString("test_2"));
executor.submit(() -> testObject.setTestString("test_3"));
executor.submit(() -> testObject.setTestString("test_4")); And what about the result? // the result might be anything!
// test_1 or test_2
// or test_3
System.out.println(testObject.getTestString()); . They are immutable classes and are implicitly classes, which means they can’t be extended. Just use a record class instead of a class: Record classes are special classes that act as transparent carriers for immutable data final record MainObject(int x) { } The compiler implicitly provides the constructor, getters, & , and . equals hashCode toString public final class MainObject {

  private final int x;

  public Rectangle(int x) {
        this.x = x;
  }

  double x() { return this.x; }
  // Implementation of equals() and hashCode(), which specify
  // that two record objects are equal if they
  // are of the same type and contain equal field values.
  public boolean equals...
  public int hashCode...
  // An implementation of toString() that returns a string
  // representation of all the record class's fields,
  // including their names.
  public String toString() {...}
} If we need to share the class now, we can do it without synchronization. That’s it for the first part! Thank you, and good luck! Notes Java primitives https://docs.oracle.com/javase/tutorial/java/nutsandbolts/datatypes.html Class object https://docs.oracle.com/javase/7/docs/api/java/lang/Object.html How we can work with collections after Java 7: https://docs.oracle.com/javase/8/docs/technotes/guides/collections/overview.html Java intermediate and terminal operation: https://javaconceptoftheday.com/java-8-stream-intermediate-and-terminal-operations/

Java Essentials: A Quick Guide for Refreshing Developers' Knowledge

Kafka is widely used for building event-driven architectures, allowing services to communicate through messages. However, when working with a foreign provider that adds a " " header to messages, you may encounter serialization and deserialization issues. microservices TypeId This article will guide you through the process of configuring a custom type mapper in Spring Kafka to handle the mapping of types between the producer and consumer services. By implementing this approach, you can successfully consume messages with different package structures and types, avoiding exceptions caused by incompatible types and ensuring smooth message processing. Understanding the deserialization Exception with " " Header TypeId In scenarios where the classes used in the consumer service have a different package structure compared to the producer service's DTO classes, the serialization exception can occur due to the mismatched class paths. DTO The serialization process relies on both the producer and the consumer having access to the same class definitions. When the consumer receives a message with a " " header referencing a class that resides in a different package than the consumer's DTO classes, the deserialization process fails to locate the appropriate class definition. This results in the "ClassNotFoundException" or "IllegalArgumentException" within the serialization exception. TypeId These exceptions can disrupt the smooth consumption of messages. Understanding the root cause of this exception is key to effectively handling the situation. An example of the serialization exception is as follows: javaCopy codejava.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer
	at org.springframework.kafka.listener.DefaultErrorHandler.handleOtherException(DefaultErrorHandler.java:198) ~[spring-kafka-3.0.7.jar:3.0.7]
Caused by: org.apache.kafka.common.errors.RecordDeserializationException: Error deserializing key/value for partition consumer-test-18 at offset 1. If needed, please seek past the record to continue consumption.
	at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1438) ~[kafka-clients-3.4.0.jar:na]
Caused by: java.lang.IllegalArgumentException: The class 'org.example.dto.out.OutMessageImplTwo' is not in the trusted packages: [java.util, java.lang, org.example.dto.in, org.example.dto.in.*]. 
If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).
	at org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:129) ~[spring-kafka-3.0.7.jar:3.0.7] The exception message provides insights into the underlying problem. In this case, the exception occurs because the class is not present in the list of trusted packages. The trusted packages serve as a safeguard to ensure secure deserialization and prevent unauthorized classes from being deserialized. org.example.dto.out.OutMessageImplTwo When deserializing a message, the default behavior of Spring Kafka is to check if the class specified in the "TypeId" header is within the trusted packages. If it is not found, an exception is thrown. This mechanism is in place to protect against potential security risks, as deserializing untrusted classes can lead to code execution vulnerabilities. To resolve this issue, you need to ensure that the consumer can properly map the incoming message to the corresponding class, even if they have different package structures. One approach is to configure a custom type mapper that explicitly handles the type mapping between the " " header and the consumer's DTO classes. TypeId By implementing a custom type mapper, you can instruct the consumer to map the " " header to the appropriate class or even ignore this header. This enables successful deserialization and ensures that the consumer can process the message correctly. TypeId In the upcoming sections, we will delve into the details of configuring and implementing a custom type mapper in Kafka listeners, allowing seamless message processing and addressing the challenges encountered when the consumer's DTO classes differ from those of the producer. Implementation of custom-type mapper In our example, we will send an implementation of OutMessageAbstract to Kafka and will read it as an implementation of the InMessageAbstract class. These classes are absolutely identical, they differ in name and package. The figure below shows the project structure. Our KafkaListener looks like this. It implements the logic for processing each type of message from the topic. @Component
@Log
@KafkaListener(
        topics = "${spring.cloud.stream.bindings.topic-in-0.destination}",
        groupId = "${spring.cloud.stream.bindings.topic-in-0.group}",
        containerFactory = "kafkaListenerContainerFactory")
public class EventListeners {
    @KafkaHandler
    public void handleInMessageImplEvent(@Payload InMessageImpl event) {
        System.out.println(event);
    }

    @KafkaHandler
    public void handleInMessageImplTwoEvent(@Payload InMessageImplTwo event) {
        System.out.println(event);
    }
} We use the kafkaListenerContainerFactory bean to configure our listener. This factory is described in KafkaConsumerConfiguration and looks like this: @Configuration
public class KafkaConsumerConfiguration {
    private final String SERVER;
    private final String SERVER_PORT;

    public KafkaConsumerConfiguration(
            @Value("${spring.cloud.stream.kafka.binder.brokers}") String server,
            @Value("${spring.cloud.stream.kafka.binder.defaultBrokerPort}") String port
    ) {
        this.SERVER = server;
        this.SERVER_PORT = port;
    }

    private <T> ConsumerFactory<String, T> typeConsumerFactory(Class<T> clazz) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, String.format("%s:%s", SERVER, SERVER_PORT));

        return new DefaultKafkaConsumerFactory<>(
                props,
                new StringDeserializer(),
                new JsonDeserializer<>(clazz));
    }

    private <T> ConcurrentKafkaListenerContainerFactory<String, T> initFactory(Class<T> clazz) {
        ConcurrentKafkaListenerContainerFactory<String, T> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(typeConsumerFactory(clazz));
        return factory;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, ?> kafkaListenerContainerFactory() {
        return initFactory(InMessageAbstract.class);
    }
} Because we're adding a “Typeld” header to the message, we'll get the error described above when we try to deserialize. However, we can configure the JsonDeserializer manually. To do this, we must define our implementation of TypeMapper. The following shows how you can define your own TypeMappers. class CustomTypeMapper<T> extends DefaultJackson2JavaTypeMapper {
    private final Class<T> clazz;

    public CustomTypeMapper(Class<T> clazz) {
        this.clazz = clazz;
    }

    @Override
    public JavaType toJavaType(Headers headers) {
        try {
            return TypeFactory.defaultInstance()
                    .constructType(ClassUtils.forName(clazz.getName(), getClassLoader()));
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
} If we add our TypeMapper implementation to the JsonDeserializer when we create the factory, then we can force the Json to be deserialized to the class we need, ignoring the header. The corrected method would look like this: private <T> ConsumerFactory<String, T> typeConsumerFactory(Class<T> clazz) {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, String.format("%s:%s", SERVER, SERVER_PORT));

    JsonDeserializer<T> jsonDeserializer = new JsonDeserializer<>(clazz);
    jsonDeserializer.setTypeMapper(new CustomTypeMapper<>(clazz));

    return new DefaultKafkaConsumerFactory<>(
            props,
            new StringDeserializer(),
            jsonDeserializer);
} Conclusion By utilizing a custom type mapper in Spring Kafka, you can overcome the challenges associated with consuming messages from foreign providers that include a " " header. This article has provided a step-by-step guide to configuring and implementing the custom type mapper, enabling smooth message processing and avoiding serialization and deserialization exceptions. With this knowledge, you can confidently integrate with external systems and consume Kafka messages with different package structures. TypeId The full project code is available on GitHub: https://github.com/stepanovD/kafka-templates/tree/master/consumer-custom-type-mapper

Handling Custom Type Mapping in Kafka Listeners for Messages with "TypeId" Header

As Staff Engineer on Lob’s team, I was recently tasked with standing up an endpoint for customers that supplied them with all the data we had stored for a particular zip code. We use pretty heavily to store our data so this boiled down to querying ES by zip code and writing the data to Amazon S3. But both services have subtle limitations on how you read/write them: ES caps your result set to 10,000 records, and Amazon S3 insists that you write to your buckets in chunks of at least 4 Megabytes. This turned out to be a perfect use case for Elixir streams! Address Verification Elasticsearch Elixir Streams Before we dive into the nitty gritty of Elasticsearch and AWS S3, let’s first do a brief refresher on Elixir Streams. What is a Stream? . This is typically accomplished by maintaining a little bit of state that describes where you currently are in your enumerable and a method of computing the next item. A trivial example would be a Stream of natural numbers: A Stream is a composable enumerable that is computed lazily Stream.iterate(0, fn x -> x + 1 end) The gory details of how this is implemented under the hood is the topic for another day. You rarely need to get that low level — the Stream library has a wealth of functions similar to that you can use as building blocks to construct your own Streams. And several Elixir library functions natively return streams: will turn a file (or other source) into a Stream allowing you to process very large files without having to read them into memory first in their entirety. The lazy aspect of Streams means that the minimum amount of the Stream is computed that’s needed to generate your answer. Enum IO.stream Imagine we want to find the 100th natural number that is palindromic and also divisible by 3: 1 
|> Stream.iterate(fn n -> n + 1 end) 
|> Stream.filter(fn n -> s = Integer.to_string(n); s == String.reverse(s) end)
|> Stream.filter(fn n -> rem(n, 3) == 0 end)
|> Enum.at(99) What I find elegant (and efficient) about this solution, is that no number beyond the answer, 20202, will be computed. Neat! If you’d like to dive in deeper, there’s a great Getting Started section on on the Elixir language website. The is also chock full of instructive examples. Enumerables and Streams Elixir API documentation Elasticsearch Now onto . The first thing to know is that After scrutinizing the documentation, it became clear that the original method that popped up on my Stack Overflow ( ) had been deprecated and the new approved method was to use . The process is: Elasticsearch ES caps the size of the result set at 10k items. Scroll API PointInTime IDs PIT IDs essentially give you a snapshot of your index at one point in time so that you get a consistent view of your search results across multiple queries. acquire a PIT for your index perform your queries using the PIT (without the index — it’s now implicit in the PID) delete the PIT That last step is crucial — PITs are extremely resource hungry so it is vital that you delete them as soon as you’re done. I borked our staging ES in my early experiments with PITs because I wasn’t deleting them and the expiration parameter seemed to have no effect. The other thing we’d like to do is abstract away the underlying PIT and paging mechanism for the end user so we provide an interface that just takes an ES query and generates a Stream of hits. Let’s start by creating and deleting the PID. is just a matter of requesting a PID associated with the index that we’re interested in running queries against. Creating the PID @spec create(ES.index_t()) :: {:ok, pit_t()} | {:error, ES.response_t()}
def create(index) do
  :post
  |> HTTP.request(url(index), HTTP.no_payload(), HTTP.json_encoding())
  |> HTTP.on_200(& &1["id"])
end is a simple HTTP delete except that the PIDs are huge so you need to supply them in the body and not as URL params. Some HTTP libraries won’t let you do this. The HTTP spec is a little muddy on the matter and last I checked on Stack Overflow there was a healthy debate on the subject. This is why we use HTTPoison — it allows payloads on DELETE requests. Deleting a PID @spec delete(pit_t()) :: :ok | {:error, HTTPoison.AsyncResponse | HTTPoison.MaybeRedirect | HTTPoison.Response}
def delete(pit) do
  url = HTTP.generate_url("_pit", ES.no_index(), HTTP.no_options())

  with {:ok, payload} <- Poison.encode(%{"id" => pit}),
        %HTTPoison.Response{status_code: 200} <-
          HTTPoison.request!(:delete, url, payload, HTTP.headers(HTTP.json_encoding())) do
    :ok
  else
    error -> {:error, error}
  end
end Our basic query is enhanced with three extra parameters: Now that the PIDs are sorted out, our next order of business is figuring out how to leverage them in our queries. size of 10,000 (the maximum allowed by elastic search) pit, a hash that contains %{id: <id>, keep_alive: “1m”} sort, a hash that contains %{_shard_doc: “desc”} (For sort, you need to provide something and is baked into every Elastic search index so that’s nice and portable.) _shard_doc @spec initial(PIT.pit_t(), ES.query_t()) :: ES.query_t()
def initial(pit, query) do
  %{
    query: query,
    pit: %{id: pit, keep_alive: PIT.expiration()},
    size: ES.max_results(),
    sort: %{"_shard_doc" => "desc"}
  }
end The trick here is that we feed into the next query the value of our sort field from the last hit of the previous result, like so: With our basic query down, we can focus on how to generate the sequence of queries that will extract all the items that satisfy our search criteria. @spec update(ES.query_t(), ES.response_t()) :: ES.query_t()
def update(query, %Req.Response{body: %{"hits" => %{"hits" => hits}}, status: 200}) do
  Map.put(query, :search_after, List.last(hits)["sort"])
end Armed with these two query-generating functions, we can code up that will pull all the results out of our desired index: Stream.iterator @spec streamer(ES.query_t()) :: {:ok, Enumerable.t()} | {:error, any}
def streamer(initial) do
  {:ok,
    initial
    |> search()
    |> Stream.iterate(fn previous -> search(Query.update(initial, previous)) end)
   }
rescue
  error -> {:error, error}
end We need a high-level function that takes the results from this function and produces a Stream of hits. But there’s a wrinkle — we have to delete the PIT when we are done with the Stream. But how will we know? The solution is to pass in a consumer that consumes the Stream and then we delete the PIT afterward, like so: @spec stream_many(ES.query_t(), ES.index_t(), ES.consumer_t(), non_neg_integer()) :: {:error, any()} | {:ok, Enumerable.t()}
def stream_many(query, index, consumer, count) do
  case PIT.create(index) do
    {:ok, pit_id} ->
      try do
        case pit_id |> Query.initial(query) |> streamer() do
          {:ok, stream} ->
            stream
            |> Stream.flat_map(& &1.body["hits"]["hits"])
            |> Stream.take(count)
            |> then(consumer)
          error -> error
        end
      after
        PIT.delete(pit_id)
      end
    error -> error
  end
rescue
  error -> {:error, error}
end But if our query returns less than 10k results, we don’t need the whole PID/sort/search_after machinery — one query will do the trick. @spec stream_one(ES.query_t(), ES.index_t(), ES.consumer_t()) :: {:ok, any()} | {:error, any()}
def stream_one(query, index, consumer) do
  query
  |> search(index)
  |> HTTP.on_200(fn body -> consumer.(body["hits"]["hits"]) end)
end Now we need a top-level function that query the size of the result and chooses between or depending on the value, like so: stream_one stream_many @spec stream(ES.query_t(), ES.index_t(), ES.consumer_t()) :: {:ok, any()} | {:error, any()}
def stream(query, index, consumer) do
  case count(query, index) do
    {:ok, count} ->
      if count > ES.max_results() do
        stream_many(query, index, consumer, count)
      else
        stream_one(query, index, consumer)
      end
    error ->
      error
  end
end AWS S3 The library we use to access is It handles all the low-level details for us but it does have one requirement: . Amazon S3 ex_aws_s3. the input data stream must be chunks of at least 4 Mbytes To accomplish this, we use the Stream function . This takes four inputs: chunk_while the enumerable to be chunked the initial value of the accumulator the step function a finalizer function Our first decision is, what should the accumulator look like that gets passed forward by the step function? It obviously should contain the list of the items in the current chunk. But more subtly, it should also contain the size of the current chunk so we don’t waste resources recomputing it each time. So that gives us a two-element tuple containing a list and an integer. Next, we turn our attention to the function. It should check if the current size is greater than or equal to the desired chunk size. If it is, we should take the list of items from the accumulator and convert them into a chunk using ; if not, we should add it to the current chunk (and update the size) using step convert add_chunk. What should do? Just push the item onto the front of a list and increase the value of by the current chunk’s size. add_chunk size The behaviour of depends on whether we care about the order of the items in the chunk being preserved in the output because the items in the list will be in reverse order so need to be reversed. But if we don’t care, we can skip that transformation. Putting this all together gives us: convert @spec chunker(Enumerable.t(), non_neg_integer(), boolean(), (any() -> non_neg_integer()), (Enumerable.t() -> any())) :: Enumerable.t()
def chunker(
        chunks,
        chunk_size,
        ordered \\ true,
        sizer \\ &String.length/1,
        joiner \\ &Enum.join/1
      ) do
  zero = {0, []}
  convert =
    if ordered do
      fn chunks -> chunks |> Enum.reverse() |> then(joiner) end
    else
      joiner
    end

  final = fn {_, chunks} -> {:cont, convert.(chunks), zero} end
  add_chunk = fn {size, chunks}, chunk -> {size + sizer.(chunk), [chunk | chunks]} end

  step = fn chunk, accum = {size, chunks} ->
    if size >= chunk_size do
      {:cont, convert.(chunks), add_chunk.(zero, chunk)}
    else
      {:cont, add_chunk.(accum, chunk)}
    end
  end

  Stream.chunk_while(chunks, zero, step, final)
end We can now combine all the components we have written to produce a simple controller action that requests an output stream of hits from our Elasticsearch section into 4M chunks that will satisfy the requirement of the ExAWS.S3 module: defmodule Controller do
  @four_megabytes 4*1024*1024

  def find_all(params) do
    query(params) |> ElasticSearch.stream(params.index, &consumer(params, &1))
  end
  
  def query(params) do
    %{term: %{params.query_term => params.query_value}}
  end
  
  def consumer(params, stream) do
    stream
    |> Stream.map(&Poison.encode!(&1["_source"]))
    |> chunker(@four_megabytes, false)
    |> S3.upload(params.bucket, params.filename)
    |> ExAws.request()
  end
end And that wraps up our exploration of Elixir streams, Elasticsearch, and AWS S3. I hope you enjoyed it! By Lob Staff Engineer, Guy Argo.

A Look at Using Elixir Streams, Elasticsearch & AWS S3

One very important thing in the software development process that is often overlooked in the early stages of a project is API documentation. Very often there are situations when developers do not describe the current API services due to tight deadlines or insufficient motivation, they plan to take up this task in the future. As a result, as the project grows, a large number of methods, parameters, and data formats appear, the description of which requires a huge amount of time and effort. This problem is especially critical when the development team is divided into small teams, such as front-end and back-end developers. In this case, without documentation, many problems can arise during the implementation process and a lot of time is spent on meetings and discussions of already implemented methods. Knowledge is passed orally from developer to developer. One of the solutions to this problem is frameworks for the automatic generation of documentation. For example, web applications often use swagger or its derivatives to describe REST APIs. This allows you to document the interfaces by which the front-end and back-end services interact. Swagger is a really good tool, but the interaction between different parts of the system is not limited to the REST API. In the case of dividing the project into microservices and using the Event-driven architecture, the interaction between services is built using events shared through the message broker. And each microservice is handled by a separate development team. It's reasonable to consider automatically documenting the events that are exchanged between services, just as swagger describes the REST API. In this article, we will look at how to enable automatic event documentation for services implemented using Spring Boot with interaction through Apache Kafka. Apache Kafka is a distributed streaming platform that allows you to process trillions of events per day. Kafka guarantees low latency, and high throughput provides fault-tolerant publish/subscribe pipelines and allows you to process event streams. To solve our problem, there is AsyncApi. AsyncAPI is an open source initiative that seeks to improve the current state of Event-Driven Architectures (EDA). AsyncApi has several Java tools that allow you to generate documentation from code. Springwolf is my choice because it provides a UI similar to springfox. Gradle dependencies To get started, we need to add dependencies to gradle.build implementation 'org.springframework.cloud:spring-cloud-starter-stream-kafka:3.2.2'
implementation 'javax.json:javax.json-api:1.1.4'
implementation 'org.glassfish:javax.json:1.1.4'

compileOnly 'org.projectlombok:lombok'

// Provides the documentation API
implementation 'io.github.springwolf:springwolf-kafka:0.6.1'
runtimeOnly 'io.github.springwolf:springwolf-ui:0.4.0' Description of events First, we need to define the events that we will send or expect from the topic. I will create an abstract DomainEvent class that will be extended with specific events and define serialization and deserialization rules in it using Jackson annotations. @NoArgsConstructor(access = AccessLevel.PROTECTED)
@AllArgsConstructor
@Getter
@Setter(AccessLevel.PROTECTED)
@EqualsAndHashCode
@JsonTypeInfo(
        use = JsonTypeInfo.Id.NAME,
        include = JsonTypeInfo.As.EXISTING_PROPERTY,
        property = "type",
        visible = true,
        defaultImpl = EntityChangedEvent.class
)
@JsonSubTypes(value = {
        @JsonSubTypes.Type(name = EntityChangedEvent.type, value = EntityChangedEvent.class),
        @JsonSubTypes.Type(name = EntityDeletedEvent.type, value = EntityDeletedEvent.class),
        @JsonSubTypes.Type(value = EntityChangedEvent.class)
})
@JsonIgnoreProperties(ignoreUnknown = true)
public abstract class DomainEvent {

    private String id;

    @JsonSerialize(using = LocalDateTimeSerializer.class)
    @JsonDeserialize(using = LocalDateTimeDeserializer.class)
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss")
    private LocalDateTime occuredOn = LocalDateTime.now();

    public abstract String getType();

} Change event class. @NoArgsConstructor(access = AccessLevel.PROTECTED)
@Getter
@Setter(AccessLevel.PRIVATE)
@EqualsAndHashCode(callSuper = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public class EntityChangedEvent extends DomainEvent {
    public static final String type = "ENTITY_CHANGED_EVENT";

    private String title;
    private String description;
    private String code;
    @NonNull
    private String entityId;  

    @Builder
    public EntityChangedEvent(
            String id,
            @NonNull String entityId,
            @NonNull String code, 
            String title,
            String description
    ) {
        super(id, LocalDateTime.now());

        if(StringUtils.isAllBlank(title, description)) {
            throw new IllegalStateException("changes is none");
        }

        this.entityId = entityId;
        this.code = code;

        this.title = title;
        this.description = description;
    }

    @Override
    public String getType() {
        return type;
    }
} Delete event class @NoArgsConstructor(access = AccessLevel.PROTECTED)
@Getter
@Setter(AccessLevel.PRIVATE)
@EqualsAndHashCode(callSuper = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public class EntityDeletedEvent extends DomainEvent {
    public static final String type = "ENTITY_DELETED_EVENT";
    @NonNull
    private String entityId;

    @Builder
    public EntityDeletedEvent(
            String id,
            @NonNull String valueId
    ) {
        super(id, LocalDateTime.now());
        this.valueId = valueId;
    }

    @Override
    public String getType() {
        return type;
    }
} Configuring Consumers and KafkaListeners After describing the event classes, we can move on to the listener configuration. To do this, we will get the configurations class KafkaConsumerConfig, in which we will initiate Beans for the ConcurrentKafkaListenerContainerFactory. We will also define the component KafkaListeners for our factory. If there are several factories, for example, if we have several families of events that need to be divided, then for each it is necessary to create a similar class. /**
Consumer configuration class for initiate factories for event families
*/
@Configuration
public class KafkaConsumerConfig {

    private final String SERVER;
    private final String SERVER_PORT;
    private final String GROUP;

    public KafkaConsumerConfig(
            @Value("${spring.cloud.stream.bindings.service-out-0.group}") String group,
            @Value("${spring.cloud.stream.kafka.binder.brokers}") String server,
            @Value("${spring.cloud.stream.kafka.binder.defaultBrokerPort}") String port
    ) {
        this.GROUP = group;
        this.SERVER = server;
        this.SERVER_PORT = port;
    }

/*
each key will be serialized to String and each event will be serialized to JSON
*/
    private <T> ConsumerFactory<String, T> typeConsumerFactory(Class<T> clazz) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, String.format("%s:%s", SERVER, SERVER_PORT));
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP);

        return new DefaultKafkaConsumerFactory<>(
                props,
                new StringDeserializer(),
                new JsonDeserializer<>(clazz));
    }

    private <T> ConcurrentKafkaListenerContainerFactory<String, T> initFactory(Class<T> clazz) {
        ConcurrentKafkaListenerContainerFactory<String, T> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(typeConsumerFactory(clazz));
        return factory;
    }

/*
init bean factory for abstract DomainEvent. This factory accept all subclasses of DomainEvent
*/
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, DomainEvent> domainEventKafkaListenerContainerFactory() {
        return initFactory(DomainEvent.class);
    }

} KafkaListener class must be define for each factory. @Component
@Log
@KafkaListener(
        topics = "topic-test",
        groupId="group-test",
        containerFactory="domainEventKafkaListenerContainerFactory")
public class KafkaListeners {

    @KafkaHandler
    void entityChangedKafkaListener(@Payload EntityChangedEvent message) {
        log.info(String.format("KafkaHandler[EntityChanged] %s %s %s", message.getType(), message.getEntityId(), message.getId()));
    }

    @KafkaHandler
    void entityDeletedKafkaListener(@Payload EntityDeletedEvent message) {
        log.info(String.format("KafkaHandler[EntityDeleted] %s %s %s", message.getType(), message.getEntityId(), message.getId()));
    }
} springwolf configuration Finally, we can move on to the springwolf configuration. We need to set up information about the application, producers, and consumers. Consumers are configured automatically based on the KafkaListener and KafkaHandler annotations, however, producers must be enumerated manually. The whole config will look something like this. @Configuration
@EnableAsyncApi
public class AsyncApiConfiguration {
    private final String SERVER;
    private final String SERVER_PORT;
    private final String CONSUMERS_BASE_PACKAGE = "consumers.package";
    private final String CHANNEL_NAME;


    public AsyncApiConfiguration(
            @Value("${spring.cloud.stream.bindings.service-out-0.destination}") String channelName,
            @Value("${spring.cloud.stream.kafka.binder.brokers}") String server,
            @Value("${spring.cloud.stream.kafka.binder.defaultBrokerPort}") String port
    ) {
        this.CHANNEL_NAME = channelName;
        this.SERVER = server;
        this.SERVER_PORT = port;
    }

    @Bean
    public AsyncApiDocket asyncApiDocket() {
        //application info
        Info info = Info.builder()
                .version("api version")
                .title("service name")
                .description("description")
                .build();

        //kafka server info
        Server kafkaServer = Server.builder()
                .protocol("kafka")
                .url(String.format("%s:%s", SERVER, SERVER_PORT))
                .build();

        //init producer builder
        KafkaProducerData.ProducerDataBuilder producerBuilder = ProducerData.builder()
                .channelName(CHANNEL_NAME)
                .channelBinding(Map.of("kafka", new KafkaChannelBinding()))
                .operationBinding(Map.of("kafka", new KafkaOperationBinding()));

        //list of event types for documentation
        List<Class<?>> events = List.of(
                EntityChangedEvent.class,
                EntityDeletedEvent.class
        );

        //generate producers
        List<ProducerData> producers = events.stream()
                .map(c -> producerBuilder
                        .payloadType(c)
                        .build()
                ).collect(Collectors.toList());

        //build AsyncApiDocket
        return AsyncApiDocket.builder()
                .basePackage(CONSUMERS_BASE_PACKAGE) //package for search KafkaListeners
                .info(info) //application info
                .producers(producers) //producers
                .server("kafka", kafkaServer) //kafka srever info
                .build();
    }
} The interface will be available at . Documentation in JSON format can be found at /springwolf/asyncapi-ui.html /springwolf/docs Conclusion In this article, we looked at how you can set up auto-generation of documentation for an Event-Driven architecture. Do not forget to take care of this at the start of your project to reduce costs in the future. Links https://www.asyncapi.com/docs/tools https://github.com/springwolf/springwolf-core https://springwolf.github.io/docs/quickstart/

Auto-generation of Documentation for Event-driven Architecture

Flutter has been booming worldwide from the past few years. While there are many popular mobile app development technologies out there, Flutter has managed to leave its mark in the mobile application development world. In this article, we’ve curated the best Flutter app templates available on the market as of July 2020. Flutter can be the framework model for every app out there. This app development framework has taken the world by storm with its simplicity, widgets, and pixel-by-pixel based mobile UI development. Simply, the apps developed using Flutter are beautiful and feature-rich. Why Flutter? Flutter makes the app development simpler by applying a Widgets mechanism. Basically, we can create the application interface by simply stacking the Widget tree with different Widgets. The flow of Widgets across the code determines the flow of UI elements on the mobile screen. It makes use of the Dart programming language at its core. Moreover, the learning curve of Flutter and Dart programming language is not that steep. It easy in comparison with other application development techs. It delivers a cross-platform application development functionality. This means the same coding implementation can be applied to create both Android and iOS-based applications. It offers many libraries out of the box. Not to mention that a plethora of libraries are out there which a Flutter developer can use to easily implement wonderful mobile app features. The templates can act as a learning material for any Flutter developer especially the beginners. With this in mind, we present you the 25 best open-source, as well as premium, Flutter app templates out there for you to explore. 1. Best Flutter App UI Templates Here is a GitHub repository offering the fully functional Flutter templates. The templates are open-source. Hence, everyone can use it freely. The templates include the design implementation for hotel booking, Fitness, design course, and custom drawer. These templates can help any Flutter developer especially the beginners to learn and get started with their own custom app. You can choose between the hotel booking UI or Fitness app UI. The features like card list view with beautiful UI and designs are intuitive and easy to use. One looking for booking apps, fitness apps or course app in Flutter environment should definitely try these templates out. 2. Flutter Chat App This template offers a complete chat app feature written entirely in Flutter and Dart. The app theme is integrated with Firebase backend. It delivers standard chat app features like real-time conversations, push notifications as well as group chat and video message functionality. It provides 30+ screens with beautiful and subtle chat app UI designs. Our favorite Flutter app theme includes the integration of camera, online status indicator, profile screen options, social media logins, custom loaders, etc. It also houses the modern Android OS functionality such as dark mode. 3 . Flutter Dating App Template This is one of the best premium Flutter app templates out there which was closely based on the Tinder app. It offers a fully functional dating app template with Firebase server backend integration. This template is entirely written in Flutter. The UI flow and designs are inspired by the Tinder UI kit. This template supports the match-making algorithm along with notifications and real-time messaging. The design implementation supports both Android and iOS platforms. Inspired by the Tinder app, this app also incorporates the swipe mechanism along with date recommendations. The features like matches, real-time conversations, dating/personal profiles, localizations are a must-have in any dating app. Thus, this template takes an extra effort to provide these features with optimized performance. The new Android features like dark-mode support is also included. 4. Flutter WordPress App Template This app template is completely written in Flutter and Dart, providing native experience and great performance. It offers optimized collection views, cross-platform support, and modular code. The UI of this template exhibits beautiful newsreader designs. The mobile template is compatible with the integration of any WordPress website. This is one of the best Flutter app templates available on the market. Users can specify the WordPress URL and the app will automatically fetch the article. Moreover, the UI implementations include an automatic slide menu with WordPress categories, articles pagination / batching, for great performance, news list, and pull to refresh functionality. Thus, this template also houses real-time updates which automatically fetches the latest articles on the list. 5. Grocery, Food, Pharmacy, Store Delivery Mobile App with Admin Panel The template is a premium tier template available in the Codecanyon marketplace for the price of $79. This mobile app template delivers market directory management system using a Flutter development environment. It offers Google Maps for showing restaurants on the maps with direction to them. 6. News Buzz New Buzz repository available freely in GitHub delivers a News App template created in Flutter. It uses News API for fetching real-time data and Firebase as the backend and authenticator. The features includes custom news feed based on selected sources, saving articles, explore news based on categories, search for topics, and sharing articles. Hence, this may be the best solution for creating your very own Flutter news app. 7. Flutter Music Player Flutter Music Player is the first open-source Flutter based material design music player with audio plugin to play local music files. The app demo is available in . As of now, the template features only support the Android platform with options for iOS features in the future. The feature integrated for Android platform include fetch local songs (Inbuilt Permission), beautiful UI with multiple themes, animation, full-fledged example, play (local files), stop, pause, seek, shuffle, album art, incomplete, onDuration / onCurrentPosition functionalities. Play Store Moreover, this project also has a plugin to fetch songs and use an audio player that can be used in any of your projects. Thus, this template might be the best solution if you are keen on building your very own music player using Flutter. 8. Fluttergram As the name suggests, the Fluttergram template is an open-source template that provides a fully-functional Instagram clone written in Flutter using Firebase / Firestore. The features are similar to those of the Instagram app. The template offers custom photo feed based on who you follow (using firebase cloud functions), post photo from camera or gallery, like posts, comment on posts, view all comments on a post, search for users, and much more. Thus, if you are looking to implement your own social media app like Instagram, then this template may be the way to get it all started. 9. BookSearch BookSearch app template is an open-source app template related to book collecting and reading. The implementation is totally based on Flutter and Dart. It offers four views for book listing, collecting, searching, and reading. Moreover, there is the feature of searching as well as bookmarking your favorite books. If you are looking to create your very own books aggregator and reading app then this app template may be the best option for you. 10. Beer Me Up This app template is particularly unusual as it has the functionality of logging the beers you drink. However, this app template may be a good source for learning the new Flutter UI, as well as customize the UI based on your needs. The template is a personal beer logging app that allows you to enter every beer you have. Hence, it makes sure to keep a log of them so that you can remember them forever. 11. Food Delivery Flutter + PHP Laravel Admin Panel Here is another premium tier Flutter app template available in the Codecanyon market place for 79 USD. The app template offers multiple Restaurants or Restaurants directory Management System. There is the feature of Google Maps for showing restaurants on the map with directions to them. It delivers entities & Models that are ready to use, clean & organized Dart Language code using MVC Pattern, easy to restyle, and theming by on your branding sites, as well as cash on delivery and payment on pickup methods. 14. Trace Trace is an open-source full-featured, modern crypto portfolio & market explorer. The features include Portfolio Overview, Aggregate Market Explorer, Portfolio Timeline, Transaction History, Portfolio Breakdown, and much more. 15. Taskist Taskist is an open-source to-do list app written completely in Flutter. The UI is simple, beautiful, and intuitive. It offers a task management system with the ability to add multiple tasks in the list format. The app uses Firebase to store the list. The UI features is based on swiping mechanism where the list can swipe right for more. There is an option of striking out the completed tasks. Moreover, it also features the ability to store and maintain tasks based on different days. 16. Fluxstore Pro – Flutter E-commerce Full App Fluxstore is a premium tier e-commerce app inspired by the Flutter framework available in the Codecanyon marketplace for $79. With the aim to reduce hours of business spent on designing, developing, and testing a mobile app, it comes as a complete solution for optimizing to deliver your app to the market with high productivity and cost-efficiency. Moreover, it offers features that satisfy all of the business requirements including e-commerce functionalities, impressive UX design, and smooth performance on both iOS and Android devices. 17. Trinity Orientation 2018 This is an app created for Trinity College at the University of Toronto. The app acts as a companion for all first-year students during orientation week. The features include college sitemap, map functionality, chat functionality, events list, gallery, etc. This app may be useful for creating similar apps for other colleges. Nonetheless, it can be used as the learning mechanism for Flutter development. 18. Deer Here is another to-do list based app template which is available freely in GitHub. This template is a minimalist Todo Planner app built around the idea of efficiency and clean aesthetic. It makes use of the BLoC (Business Logic Component) pattern to manage the app state. It showcases 4 screens. The first one is the list entry screen where you enter the to-do task, then the list screen to view the to-do task list, the screen for task list detail, and the screen to edit the to-do task. Thus, this template can be the right choice to develop your very own to-do list app using Flutter in the BLoC pattern. 19. TailorMade TailorMade is an open-source daily routine management app template made especially for Fashion designers. It started out as an experiment with Flutter, ReBLoC, and but later turned into a valuable tool for managing a Fashion designer’s daily routine. Firebase Cloud Functions Moreover, the template is simple, clean, and overall has a very smooth intuitive feel. This template offers 20+ screens which are all fully functional. Hence, it is highly suggested to check this template out as you can have new features to learn in Flutter and also develop your very own routine management app with some tweaks. 20. instory instory is an open-source app available in Github free of cost. This mobile template delivers a Flutter app with beautiful UI features with 3 screens. The major objective of this app template is to download Instagram stories. Moreover, the app offers the search functionality to search for a particular profile story and download them. 21. StreamApp – Streaming Movies, TV Series and Live TV Stream app is another premium tier app available in the CodeCanyon store for $29. It delivers features for streaming movies, TV series, and live TV. This app template has support for formats on external servers, local or AWS S3 storage. It saves a lot of effort for the administrator with features such as remote configuration, auto-import of information from TheMovieDB, integration with AWS API, Firebase, AdMob, and more. Moreover, the admin panel built using Vue.js and backed by Laravel guarantees you the full power of new technologies. 22. Reply  Flutter This Flutter app template delivers a pixel-perfect implementation of the in . is an email app that makes use of Material Design components and Material Theming to implement an on-brand communication experience. Reply Material Design Case Study Flutter Reply Material Design Case Study It mainly focuses on UI elements of Flutter such as animation, layouts, and transitions. This template aims to showcase how powerful Flutter has become as a UI framework. Hence, this template can be used as a learning mechanism for Flutter app development. 23. Enigma Enigma is an open-source one-to-one chat app developed entirely in Flutter. The backend is based on Firebase implementation. This app template delivers real-time one-to-one chatting experience with minimal chatting features. Since this is a Firebase dependent project, it uses Firebase Phone Authentication, Cloud Firestore, Firebase Storage, and Firebase In-App Messaging. Moreover, the chat interfaces are passcode protected. This template can be a medium to learn how to develop a Flutter chatting app using Firebase features. Hence, you can also develop your very own Flutter chatting app using this template as a starter package. 24. Flutter Pokédex How can we forget one of the legendary anime Pokémon? There has been a number of games based on it as well. Those who watch Pokémon or play their games simply cannot ignore the vital use of the Pokédex. Well, this template offers the functionality of a Pokédex and is completely built using Flutter. The app hosts the Pokémon list along with their data. Moreover, what makes this template great is the Widget combination, animations, and charts used in it. This can be a great app to learn the use of animations and different UI components in Flutter. Even if you are not a fan of Pokémon, you should still check this template out. Conclusion With this plethora of Flutter app templates available out there, one can surely develop the app they are looking for. Most app templates are open-source which means they are free to use as the starter template for creating your very own Flutter application. Some premium tier templates are available which provide powerful features that you can mildly customize based on your needs and deploy to the market. The best Flutter app templates can be the breeding ground for many application startups and solutions for companies. Thus, it is highly suggested to check each one of these flutter app templates for a better understanding of the Flutter app market and its use cases. Did we miss your favorite Flutter app template? Let us know in the comments! Previously published at https://www.instaflutter.com/design/flutter-app-templates-examples-best-25-in-2020/

Top 25 Flutter Mobile App Templates in 2020

Adding. Changing. Hacking flavours. After 3 years of developing, I’ve begun publicly marketing . Streamplate Over time it’s shifted in design, scope and intent, and now stands as an operating-like-system for consumers and merchants alike to manage their eating and drinking needs. At its core, Streamplate offers a personalised assortment of merchants with a sorted listing of menu-items to cater for each user’s needs. While the first iteration of the Alpha app has been completed, there’s still much to be done. Presently, each computation of the merchant listings takes into account approximately a few thousand-variables. What I’ve realised over the past few months is that while having such a high-dimensional space isn’t anything unique, the frequency of these calculations is. Eating and drinking isn’t based on “universal truths” which sets it apart from the paradigms that guide Google’s search engine or Spotifys’ playlists curations. What someone wants to order is a function of their perspective in that moment in time. When we break down what a “perspective” is in factor-analysis, a high-level structure emerges: Perception => State of values, Consciousness => Stream of perceptions, Levels of consciousness => Attentiveness to rates of changing intensities of consciousness. A key-insight here is how underlying any listing is a measurement of “values”, hence the need for a broad-spectrum of values and hence an increasingly-growing dimensional-space. What this means is that Streamplate needs to consistently be measuring the pulse of a user’s perception so when they do call for a listing, we can be sure to accurately match their needs. Think of how varying your tastes can be, particularly under duresses like unforgiving summer heat or compelling social influence. Reaching this understanding required a progressive set of design iterations, each time further investigating the guiding principles affecting our users’ decisions. As I continue working on the back-end code to implement these understandings, I’d like to briefly share some moments in the design process. Jump to the front of the queue for a year by signing up for Streamplate’s ! Limited Release

Designing Streamplate

Using Arrow Flight SQL Protocol in Apache Doris 2.1 For Super Fast Data Transfer

Real-Time tricks: Harnessing Kafka Streams for Seamless Data Tasks

Python Bokeh is one of the best Python packages for data visualization. Today, we are going to see some Python Bokeh Examples. I have also provided the Python Bokeh project source code on GitHub. Learn this easy visualization tool and add it to your Python stack. What is Python Bokeh? Python is a data visualization tool, or we can also say Python Bokeh is used to plot various types of graphs. There are various other graph plotting libraries like matplotlib, but Python Bokeh graphs are dynamic in nature, which means you can interact with the generated graph. See the below examples… Bokeh Installation 💻: Python Bokeh can be easily installed using PIP. You can install the Python Bokeh easily by running the command: pip install bokeh Now everything is ready, let’s go through the examples 🏃‍♂️… 1. LinePlot from bokeh.plotting import figure, show, output_notebook
x = [1, 2, 3, 4, 5]
y = [4, 5, 5, 7, 3]
p = figure(title="{LinePlot Python Bokeh Example")
p.line(x, y, line_width=2)
output_notebook()
show(p) | | Live Preview Source Code 🌿 Contribute 2. Scatter Plot from bokeh.plotting import figure, show, output_notebook
x = [1, 2, 3, 4, 5]
y = [4, 5, 5, 7, 3]
p = figure(title="Scatter Plot Python Bokeh Example by PratikPathak.com")
p.circle(x, y, size=10, color="navy", alpha=0.5)
output_notebook()
show(p) 3. Bar Chart from bokeh.plotting import figure, show, output_notebook
categories = ["A", "B", "C", "D", "E"]
values = [10, 15, 8, 12, 6]
p = figure(x_range=categories, title="Bar Chart Python Bokeh Example by PratikPathak.com")
p.vbar(x=categories, top=values, width=0.9)
output_notebook()
show(p) 4. Histogram from bokeh.plotting import figure, show, output_notebook
import numpy as np
data = np.random.normal(0, 1, 1000)
p = figure(title="Histogram Python Bokeh Example by PratikPathak.com")
p.hist(data, bins=30, color="navy", alpha=0.5)
output_notebook()
show(p) 5. Pie Chart from bokeh.plotting import figure, show, output_notebook
labels = ["A", "B", "C", "D"]
values = [10, 15, 8, 12]
p = figure(title="Pie Chart Python Bokeh Example by PratikPathak.com")
p.wedge(x=0, y=0, radius=0.4, start_angle=0.6, end_angle=2.6, color=["red", "green", "blue", "yellow"], legend_label=labels)
output_notebook()
show(p) | | Live Preview Source Code 🌿 Contribute 6. Time Series Plot from bokeh.plotting import figure, show, output_notebook
from datetime import datetime, timedelta
start = datetime(2023, 1, 1)
end = start + timedelta(days=30)
x = [start + timedelta(days=i) for i in range((end-start).days)]
y = [10, 15, 8, 12, 6, 18, 9, 14, 7, 11, 5, 16, 8, 13, 6]
p = figure(x_axis_type="datetime", title="Time Series Plot Python Bokeh Example by PratikPathak.com")
p.line(x, y, line_width=2)
output_notebook()
show(p) 7. Linked Brushing from bokeh.plotting import figure, show, output_notebook
from bokeh.models import ColumnDataSource
x1 = [1, 2, 3, 4, 5]
y1 = [4, 5, 5, 7, 3]
x2 = [2, 3, 4, 5, 6]
y2 = [2, 4, 6, 8, 4]
source = ColumnDataSource(data=dict(x1=x1, y1=y1, x2=x2, y2=y2))
p1 = figure(title="Scatter Plot 1")
p1.circle('x1', 'y1', source=source)
p2 = figure(title="Scatter Plot 2 Python Bokeh Example by PratikPathak.com")
p2.circle('x2', 'y2', source=source)
output_notebook()
show(p1, p2) 8. Hover Tooltips from bokeh.plotting import figure, show, output_notebook
from bokeh.models import HoverTool
x = [1, 2, 3, 4, 5]
y = [4, 5, 5, 7, 3]
p = figure(title="Hover Tooltips Python Bokeh Example by PratikPathak.com")
p.circle(x, y, size=15, fill_color="navy", line_color="white", alpha=0.5)
hover = HoverTool(tooltips=[("(x,y)", "(@x, @y)")])
p.add_tools(hover)
output_notebook()
show(p) 9. Annotations from bokeh.plotting import figure, show, output_notebook
from bokeh.models import Arrow, VectorRenderer, Label
x = [1, 2, 3, 4, 5]
y = [4, 5, 5, 7, 3]
p = figure(title="Annotations Python Bokeh Example by PratikPathak.com")
p.circle(x, y, size=15, fill_color="navy", line_color="white", alpha=0.5)
arrow = Arrow(x_start=2, y_start=4, x_end=3, y_end=5, line_width=2, line_color="red")
label = Label(x=3, y=6, text="This is a label", render_mode='css', border_line_color='black', border_line_alpha=1.0, background_fill_color='white', background_fill_alpha=0.5)
p.add_layout(arrow)
p.add_layout(label)
output_notebook()
show(p) 10. Custom Glyphs from bokeh.plotting import figure, show, output_notebook
from bokeh.models.glyphs import Asterisk
x = [1, 2, 3, 4, 5]
y = [4, 5, 5, 7, 3]
p = figure(title="Custom Glyphs Python Bokeh Example by PratikPathak.com")
p.add_glyph(x, y, Asterisk(size=20, line_color="red", fill_color="yellow"))
output_notebook()
show(p) | | Live Preview Source Code 🌿 Contribute 11. Gridlines and Axes from bokeh.plotting import figure, show, output_notebook
x = [1, 2, 3, 4, 5]
y = [4, 5, 5, 7, 3]
p = figure(title="Gridlines and Axes Python Bokeh Example by PratikPathak.com", x_range=(0, 6), y_range=(0, 8))
p.grid.grid_line_color = "grey"
p.grid.grid_line_dash = [6, 4]
p.xaxis.axis_label = "X-axis"
p.yaxis.axis_label = "Y-axis"
p.line(x, y, line_width=2)
output_notebook()
show(p) 12. Legend from bokeh.plotting import figure, show, output_notebook
x1 = [1, 2, 3, 4, 5]
y1 = [4, 5, 5, 7, 3]
x2 = [2, 3, 4, 5, 6]
y2 = [2, 4, 6, 8, 4]
p = figure(title="Legend Python Bokeh Example by PratikPathak.com")
p.line(x1, y1, line_width=2, color="red", legend_label="Line 1")
p.line(x2, y2, line_width=2, color="blue", legend_label="Line 2")
p.legend.location = "top_left"
output_notebook()
show(p) 13. Categorical Plots from bokeh.plotting import figure, show, output_notebook
fruits = ["Apples", "Pears", "Nectarines", "Plums", "Grapes", "Strawberries"]
counts = [5, 3, 4, 2, 4, 6]
p = figure(x_range=fruits, title="Categorical Plots Python Bokeh Example by PratikPathak.com")
p.vbar(x=fruits, top=counts, width=0.9)
p.xaxis.axis_label = "Fruit"
p.yaxis.axis_label = "Count"
p.xaxis.axis_label_text_font_size = "12pt"
p.yaxis.axis_label_text_font_size = "12pt"
output_notebook()
show(p) 14. Subplots from bokeh.plotting import figure, show, output_notebook
from bokeh.layouts import gridplot
x1 = [1, 2, 3, 4, 5]
y1 = [4, 5, 5, 7, 3]
x2 = [2, 3, 4, 5, 6]
y2 = [2, 4, 6, 8, 4]
p1 = figure(title="Scatter Plot 1 Python Bokeh Example by PratikPathak.com")
p1.circle(x1, y1, size=10, color="navy", alpha=0.5)
p2 = figure(title="Scatter Plot 2 Python Bokeh Example by PratikPathak.com")
p2.circle(x2, y2, size=10, color="red", alpha=0.5)
grid = gridplot([[p1, p2]], plot_width=400, plot_height=400)
output_notebook()
show(grid) 15. Interactive Plots from bokeh.plotting import figure, show, output_notebook
from bokeh.models import ColumnDataSource, HoverTool, BoxSelectTool
import numpy as np
x = np.random.random(1000)
y = np.random.random(1000)
source = ColumnDataSource(data=dict(x=x, y=y))
p = figure(title="Interactive Plots Python Bokeh Example by PratikPathak.com", tools="hover,box_select")
p.circle('x', 'y', source=source, size=3, color="navy", alpha=0.5)
hover = HoverTool(tooltips=[("(x,y)", "(@x, @y)")])
p.add_tools(hover)
output_notebook()
show(p) | | Live Preview Source Code 🌿 Contribute 16. Linked Panning and Zooming from bokeh.plotting import figure, show, output_notebook
from bokeh.models import Range1d
x1 = [1, 2, 3, 4, 5]
y1 = [4, 5, 5, 7, 3]
x2 = [2, 3, 4, 5, 6]
y2 = [2, 4, 6, 8, 4]
p1 = figure(title="Scatter Plot 1 Python Bokeh Example by PratikPathak.com", x_range=Range1d(0, 6), y_range=Range1d(0, 8))
p1.circle(x1, y1, size=10, color="navy", alpha=0.5)
p2 = figure(title="Scatter Plot 2 Python Bokeh Example by PratikPathak.com", x_range=p1.x_range, y_range=p1.y_range)
p2.circle(x2, y2, size=10, color="red", alpha=0.5)
output_notebook()
show(p1, p2) 17. Linked Axis Ranges from bokeh.plotting import figure, show, output_notebook
from bokeh.models import Range1d
x1 = [1, 2, 3, 4, 5]
y1 = [4, 5, 5, 7, 3]
x2 = [2, 3, 4, 5, 6]
y2 = [2, 4, 6, 8, 4]
p1 = figure(title="Scatter Plot 1", x_range=(0, 6), y_range=(0, 8))
p1.circle(x1, y1, size=10, color="navy", alpha=0.5)
p2 = figure(title="Scatter Plot 2", x_range=p1.x_range, y_range=p1.y_range)
p2.circle(x2, y2, size=10, color="red", alpha=0.5)
output_notebook()
show(p1, p2) 18. DateTime Axis from bokeh.plotting import figure, show, output_notebook
from datetime import datetime, timedelta
start = datetime(2023, 1, 1)
end = start + timedelta(days=30)
x = [start + timedelta(days=i) for i in range((end-start).days)]
y = [10, 15, 8, 12, 6, 18, 9, 14, 7, 11, 5, 16, 8, 13, 6]
p = figure(x_axis_type="datetime", title="Datetime Axis Python Bokeh Example by PratikPathak.com")
p.line(x, y, line_width=2)
output_notebook()
show(p) 19. Heatmap from bokeh.plotting import figure, show, output_notebook
import numpy as np
data = np.random.rand(10, 10)
p = figure(title="Heatmap Python Bokeh Example by PratikPathak.com")
p.image(image=[data], x=0, y=0, dw=10, dh=10, palette="Viridis256")
output_notebook()
show(p) 20. Streamgraph from bokeh.plotting import figure, show, output_notebook
from bokeh.models import ColumnDataSource
import numpy as np
# Generate some sample data
x = np.linspace(0, 10, 100)
y1 = np.cumsum(np.random.randn(100))
y2 = np.cumsum(np.random.randn(100))
y3 = np.cumsum(np.random.randn(100))
y4 = np.cumsum(np.random.randn(100))
y5 = np.cumsum(np.random.randn(100))
source = ColumnDataSource(data=dict(x=x, y1=y1, y2=y2, y3=y3, y4=y4, y5=y5))
p = figure(title="Streamgraph Python Bokeh Example by PratikPathak.com")
p.area(x='x', y1='y1', y2='y2', y3='y3', y4='y4', y5='y5', source=source, line_color=None)
output_notebook()
show(p) | | Live Preview Source Code 🌿 Contribute 21. Polar Plot from bokeh.plotting import figure, show, output_notebook
import numpy as np
r = np.linspace(0.1, 1, 10)
theta = np.linspace(0, 2*np.pi, 10)
p = figure(title="Polar Plot Python Bokeh Example by PratikPathak.com", plot_width=600, plot_height=600, x_range=(-1.2, 1.2), y_range=(-1.2, 1.2), toolbar_location=None)
p.polar(r, theta, color="navy", alpha=0.5, line_width=2)
output_notebook()
show(p) 22. Donut Chart from bokeh.plotting import figure, show, output_notebook
from bokeh.models import ColumnDataSource
labels = ["A", "B", "C", "D"]
values = [10, 15, 8, 12]
source = ColumnDataSource(data=dict(labels=labels, values=values))
p = figure(title="Donut Chart Python Bokeh Example by PratikPathak.com", x_range=labels)
p.wedge(x=0, y=1, radius=0.4, start_angle=0.6, end_angle=2.6, color=["red", "green", "blue", "yellow"], legend_field="labels")
p.wedge(x=0, y=1, inner_radius=0.3, outer_radius=0.4, start_angle=0.6, end_angle=2.6, color="white")
p.axis.visible = False
p.grid.grid_line_color = None
output_notebook()
show(p) 23. Treemap from bokeh.plotting import figure, show, output_notebook
from bokeh.models import TreemapRenderer, ColumnDataSource
import pandas as pd

# Sample data
data = pd.DataFrame({
    'label': ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'],
    'parent': ['', 'A', 'A', 'B', 'B', 'C', 'C', 'D'],
    'value': [10, 5, 4, 3, 2, 1, 1, 1]
})

source = ColumnDataSource(data)
p = figure(title="Treemap Python Bokeh Example by PratikPathak.com")
treemap = TreemapRenderer(
    source=source,
    label='label',
    parent='parent',
    value='value',
    color_discrete_sequence=["#e6550d", "#3182bd", "#31a354", "#756bb1", "#636363", "#9e9ac8", "#de2d26", "#fd8d3c"],
)
p.add_layout(treemap)
output_notebook()
show(p) If you get any error, make sure you have installed Python pandas by using the “pip install pandas” command 24. Network Plot from bokeh.plotting import figure, show, output_notebook
from bokeh.models import ColumnDataSource, HoverTool, NodesAndLinkedEdges
import networkx as nx

# Sample data
nodes = ['A', 'B', 'C', 'D', 'E', 'F']
edges = [('A', 'B'), ('A', 'C'), ('B', 'D'), ('B', 'E'), ('C', 'F')]
G = nx.Graph()
G.add_nodes_from(nodes)
G.add_edges_from(edges)

node_positions = nx.spring_layout(G)
node_x = [node_positions[node][0] for node in G.nodes()]
node_y = [node_positions[node][1] for node in G.nodes()]
node_color = ['red', 'green', 'blue', 'yellow', 'purple', 'orange']

source = ColumnDataSource(data=dict(
    x=node_x,
    y=node_y,
    color=node_color,
    node=nodes
))

p = figure(title="Network Plot Python Bokeh Example by PratikPathak.com", x_range=(-1.1, 1.1), y_range=(-1.1, 1.1))
p.add_tools(HoverTool(tooltips=[("Node", "@node")]))
p.circle('x', 'y', size=15, color='color', source=source)

edge_x = []
edge_y = []
for start, end in edges:
    x0, y0 = node_positions[start]
    x1, y1 = node_positions[end]
    edge_x.append(x0)
    edge_x.append(x1)
    edge_x.append(None)
    edge_y.append(y0)
    edge_y.append(y1)
    edge_y.append(None)

p.multi_line(edge_x, edge_y, color="black", alpha=0.5, line_width=1)

output_notebook()
show(p) 25. Sankey Diagram from bokeh.plotting import figure, show, output_notebook
from bokeh.models import Sankey, ColumnDataSource

# Sample data
flows = [
    ('A', 'X', 5),
    ('B', 'X', 3),
    ('A', 'Y', 2),
    ('B', 'Y', 4),
    ('X', 'C', 4),
    ('X', 'D', 3),
    ('Y', 'C', 2),
    ('Y', 'D', 4)
]

source = ColumnDataSource(data=dict(
    source=[flow[0] for flow in flows],
    target=[flow[1] for flow in flows],
    value=[flow[2] for flow in flows]
))

p = figure(plot_width=800, plot_height=400, title="Sankey Diagram Python Bokeh Example by PratikPathak.com")
sankey = Sankey(source=source, label_text_color="white", label_text_font_size="8pt")
p.renderers.append(sankey)

output_notebook()
show(p) | | Live Preview Source Code 🌿 Contribute 26. Chord Diagram from bokeh.plotting import figure, show, output_notebook
from bokeh.models import ColumnDataSource, CategoricalColorMapper
import numpy as np

# Sample data
nodes = ['A', 'B', 'C', 'D', 'E', 'F']
connections = [
    ('A', 'B', 0.5), ('A', 'C', 0.3), ('A', 'D', 0.2),
    ('B', 'C', 0.4), ('B', 'D', 0.3), ('B', 'E', 0.3),
    ('C', 'D', 0.2), ('C', 'E', 0.3), ('C', 'F', 0.5),
    ('D', 'E', 0.2), ('D', 'F', 0.3), ('E', 'F', 0.4)
]

source = ColumnDataSource(data=dict(
    start=np.array([con[0] for con in connections]),
    end=np.array([con[1] for con in connections]),
    value=np.array([con[2] for con in connections])
))

p = figure(title="Chord Diagram Python Bokeh Example by PratikPathak.com", plot_width=800, plot_height=800)
chord = p.chord(
    source=source, 
    start_angle=np.pi / 2,
    direction="clockwise",
    colors=CategoricalColorMapper(factors=nodes, palette="Spectral11")
)

output_notebook()
show(p) How to Contribute? Feel free to open a PR request on our GitHub repo. Steps to contribute: Fork the repo Make changes in the Forked repo and save Open a Pull Request That’s it 😄! 🌿 Contribute Conclusion In this Repo, I have shared 25+ Python Bokeh examples that can help you learn Python Bokeh. Feel free to contribute to our GitHub repo and keep it updated. Also published . here

Getting Started With Python Bokeh: 25+ Data Visualization Examples With Source Code

Hello everyone and thanks for dropping by. Today I would like to explore how one might go about creating their own sitemap generator script for a static Next.js website. Of course there are packages out there that do just this, but there is great value in writing your own and understanding the solution fully. I'll assume that readers have some familiarity with Next.js, JavaScript, Node.js and SEO to set the tone for what follows. In addition, I assume the reader has a static Next.js website or can create one to follow along with the post. Ok, now let's get started. Background The motivation for this post really came from some work I was doing on my personal website: . The goal of the site is to be an engaging and educational blog that is communicated in a digestible way for any reader. I wanted to customize the blog as much as I'd like with little to no cost and reacquaint myself with React.js. I initially thought a vanilla React.js application would be great, but SPAs have their limitations when it comes to SEO if you aren't server side rendering your pages. So I chose to host a static Next.js site on Google Cloud Storage to make my content indexable by search engines. afro-cloud.com Let's get refocused on the concept of sitemaps. You can think of sitemaps as a list of pages for your website that help search engines understand the contents of your website. In short, they tell search engine crawlers what contents on your site are important to you. But sitemaps aren't just limited to pages, they can be used to indicate other types of content on your site like video for example. So if you want your pages to be indexed by a search engine like Google, it's important to indicate to their crawlers what content matters to you. The sitemap.xml file conveys this information and we will be building a script to generate one for a static Next.js website. Understanding Next.js Routing Next.js has two different routing paradigms: "Using app router" and "Using page router". For the purposes of this post, we will be using the app router approach. The TLDR is that all pages for our website will be contained within a top-level folder called app/. But within app/, we can have nested directories to represent the hierarchy of pages for our site. The way we indicate to Next.js that we have pages for our website is by having a page.jsx or page.tsx file within those directories. Take this directory structure for our app represented in a nested list for example: /app page.tsx /about page.tsx /blog page.tsx /post1 page.tsx /post2 page.tsx For this example we have the following page paths: /app/page.tsx /app/about/page.tsx /app/blog/page.tsx /app/blog/post1/page.tsx /app/blog/post2/page.tsx For sake of example, let's say we're building a website with the domain: wethebest.com. If we translate the above list to client routes we get: wethebest.com wethebest.com/about wethebest.com/blog wethebest.com/blog/post1 wethebest.com/blog/post2 Since we are using a static website approach, we need to map our client side routes to the fully qualified path of the HTML files corresponding to each page to ensure refreshes navigate to the same page. Doing so yields the following: wethebest.com wethebest.com/about.html wethebest.com/blog.html wethebest.com/blog/post1.html wethebest.com/blog/post2.html Ok, so now we have some pages, but what does Next.js offer to create a sitemap.xml for our application? Next.js provides , but you have to define a configuration object that corresponds to what you want in your sitemap.xml. But we can do better. I for one am lazy and don't want to add a new entry to a configuration every time I add a page to my site. Let's create our own script to generate a sitemap for us. some support Writing a sitemap.xml Generator Script So how might you start to tackle this problem of generating our own sitemap.xml generator script? We know how the client side routing works in Next.js for our use case. We know that our sitemap.xml URL entries need to have the static HTML file included in the URL for each page. We know that whenever we add a new page to our site, we want the URL of that page to be reflected in our sitemap.xml. Hopefully you're thinking about reading the content of our app directory and using that information to build our sitemap. If not, no worries! We'll go through it together. Let's first start by creating a file at the root of our project alongside our . The contents of the file will be the following to start: sitemap.js package.json const fs = require("node:fs");
const { SitemapStream, streamToPromise } = require("sitemap");
const { Readable } = require("stream");

const SITEMAP_HOSTNAME = "https://www.wethebest.com";

(async () => {
// Create [SitemapStream] that will ultimately populate our sitemap.xml.
const stream = new SitemapStream({
hostname: SITEMAP_HOSTNAME,
});
})(); We'll be using the Node.js file system sync API to read the contents of our directories. We'll also be using an NPM package called sitemap that provides utilities to help make creating our sitemap easy. In order to take advantage of the sitemap package, we'll need to create a readable stream that Node.js provides. We have a constant variable declaration that we'll use to build our URLs for the sitemap.xml file. Finally, we have an immediately invoked function expression which will be executed whenever we execute this file. Now let's add the code to start traversing our app/ directory and find the pages for our website. The first thing we'll want to do is start exploring our top-level app/ folder and find all of our pages file paths. The reason we are doing this is that we can map the file path to the corresponding HTML files for our site (look back at the routing section if you need a refresher). We can use the following code to do that Update your sitemap.js file to look like this: const fs = require("node:fs");
const { SitemapStream, streamToPromise } = require("sitemap");
const { Readable } = require("stream");

const PAGE_FILENAME = "page.tsx";
const SITEMAP_HOSTNAME = "https://www.wethebest.com";
const SITEMAP_PATHS = [];

Populates [SITEMAP_PATHS] with the fully qualified URLs for our pages to be contained

within our sitemap.xml file.

@param {string} path The path we are currently searching.

@param {string} isRoot Flag that denotes whether or not we are processing our root path.

@return {void}
*/
const getAllUrlsForSitemap = (path, isRoot) => {
// Get all files within our current directory
const files = fs.readdirSync(path);
files.forEach((file) => {
// Not using template strings due to article format on website.
const dirPath = 'path' + '/' + 'file';
if (file.includes(PAGE_FILENAME) && !isRoot) {
SITEMAP_PATHS.push(dirPath);
} else {
try {
const stats = fs.statSync(dirPath);
if (stats.isDirectory()) {
// Continue searching until we exhaust all directories.
getAllUrlsForSitemap(dirPath);
}
} catch (err) {}
}
});
};

(async () => {
// Create [SitemapStream] that will ultimately populate our sitemap.xml.
const stream = new SitemapStream({
hostname: SITEMAP_HOSTNAME,
});

// Populate [SITEMAP_PATHS] based on our [PAGE_FILENAME].
getAllUrlsForSitemap(APP_ROOT, true);
})(); You'll notice that we have a new method declaration: getAllUrlsForSitemap. This method navigates our app/ directory contents and searches for page.tsx files as well as other directories. We have a special case for our root page (app/page.tsx) because we'll add that entry later. Since we have some nested paths, we will need to explore more than the top-level app folder. The output of the readdirSync gives us all the contents of the directory and we loop through the contents to check for our pages and nested directories. Finally, when we find a page, we add the corresponding path to a variable SITEMAP_PATHS that we will use for processing later on. Now let's tie it all together. We have all of the paths for the pages we need to generate our sitemap, now all we have to do is map those to the HTML files and create the sitemap. Update your sitemap.js file to look like this: const fs = require("node:fs");
const { SitemapStream, streamToPromise } = require("sitemap");
const { Readable } = require("stream");

const APP_ROOT = "/Users/claudioherrera/Documents/GitHub/wethebest.com/app";
const PAGE_FILENAME = "page.tsx";
const SITEMAP_HOSTNAME = "https://www.wethebest.com";
const SITEMAP_PATHS = [];
const TODAYS_DATE_AS_ARRAY = new Date().toLocaleDateString().split("/");
const WRITE_FILE_PATH = "./app/sitemap.xml";

/**

Populates [SITEMAP_PATHS] with the fully qualified URLs for our pages to be contained

within our sitemap.xml file.

@param {string} path The path we are currently searching.

@param {string} isRoot Flag that denotes whether or not we are processing our root path.

@return {void}
*/
const getAllUrlsForSitemap = (path, isRoot) => {
// Get all files within our current directory
const files = fs.readdirSync(path);
files.forEach((file) => {
// Not using template strings due to article format on website.
const dirPath = 'path' + '/' + 'file';
if (file.includes(PAGE_FILENAME) && !isRoot) {
SITEMAP_PATHS.push(dirPath);
} else {
try {
const stats = fs.statSync(dirPath);
if (stats.isDirectory()) {
// Continue searching until we exhaust all directories.
getAllUrlsForSitemap(dirPath);
}
} catch (err) {}
}
});
};

/**

Returns an object containing the url and lastmod sitemap properties.

@param {string} path The path we are currently searching.

@param {string} isRoot Flag that denotes whether or not we are processing our root path.

@return {Object} url: URL for page on website. lastmod: last modification date for our page.
*/
const mapToSitemapFormat = (path, lastmod) => ({
// Not using template strings due to article format on website.
url: path
.replace(APP_ROOT + '/', "")
.replace('/' + PAGE_FILENAME, "") + '.html',
lastmod,
});

/**

Returns a string representation of a date unit. e.g. dd, mm, yy

@param {string} unit Date unit as a number in string format. e.g. month, day, year

@param {string} isRoot Flag that denotes whether or not we are processing our root path.

@return {Object} url: URL for page on website. lastmod: last modification date for our page.
*/
const padDateUnit = (unit) => ('unit'.length < 2 ? '0' + 'unit' : unit);

(async () => {
console.info("Generating sitemap...");

// Create [SitemapStream] that will ultimately populate our sitemap.xml.
const stream = new SitemapStream({
hostname: SITEMAP_HOSTNAME,
});

// Populate [SITEMAP_PATHS] based on our [PAGE_FILENAME].
getAllUrlsForSitemap(APP_ROOT, true);

try {
// Create our sitemap.xml contents. Not using template strings due to article format on website.
const LAST_MODIFIED_DATE = TODAYS_DATE_AS_ARRAY[2] + '-' + padDateUnit(
TODAYS_DATE_AS_ARRAY[0]
) + '-' + padDateUnit(TODAYS_DATE_AS_ARRAY[1]);
const ROOT_SITEMAP_ENTRY = {
url: SITEMAP_HOSTNAME,
lastmod: LAST_MODIFIED_DATE,
};
const content = await streamToPromise(
Readable.from([
ROOT_SITEMAP_ENTRY,
...SITEMAP_PATHS.map((url) =>
mapToSitemapFormat(url, LAST_MODIFIED_DATE)
),
]).pipe(stream)
).then((data) => data.toString());

console.info("Successfully generated sitemap...");

// Write our sitemap contents to [WRITE_FILE_PATH].
fs.writeFileSync(WRITE_FILE_PATH, content);

} catch (err) {
console.error(err);
}
})(); The core of the difference between this snippet and the one above is that we have now tied together the last two pieces to generate our sitemap. We are mapping our SITEMAP_PATHS to the sitemap.xml format by creating objects that contain a url and lastmod key. These objects are then fed into the sitemap package to generate the sitemap in XML format. Finally, we write the sitemap.xml to the location of our choosing. Let's execute the sitemap.js script by running "node sitemap.js" (note you will need to have Node.js installed to be in the current working directory of the sitemap file). You should now see a sitemap.xml file in your app folder and it should look something like: <?xml version="1.0" encoding="UTF-8"?>
<urlset
xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
xmlns:news="http://www.google.com/schemas/sitemap-news/0.9"
xmlns:xhtml="http://www.w3.org/1999/xhtml"
xmlns:image="http://www.google.com/schemas/sitemap-image/1.1"
xmlns:video="http://www.google.com/schemas/sitemap-video/1.1">
<url>
<loc>https://www.wethebest.com/</loc>
<lastmod>2024-03-10T00:00:00.000Z</lastmod>
</url>
<url>
<loc>https://www.wethebest.com/about.html</loc>
<lastmod>2024-03-10T00:00:00.000Z</lastmod>
</url>
<url>
<loc>https://www.wethebest.com/blog.html</loc>
<lastmod>2024-03-10T00:00:00.000Z</lastmod>
</url>
<url>
<loc>https://www.wethebest.com/blog/post1.html</loc>
<lastmod>2024-03-10T00:00:00.000Z</lastmod>
</url>
<url>
<loc>https://www.wethebest.com/blog/post2.html</loc>
<lastmod>2024-03-10T00:00:00.000Z</lastmod>
</url>
</urlset> Nice! Now, whenever you add a new page to your Next.js app, you can easily include it in your sitemap. I highly recommend adding a new production build rule to your application to incorporate this sitemap generation whenever you deploy a new version of your website. I hope you found this post useful. Cheers! Also appears . here

Next.js Sitemap Generator Script

Retrieval-Augmented Generation (RAG) blends the generative abilities of LLMs with the retrieval of information from diverse knowledge bases (KBs). However, traditional implementations of RAG have often relied on pre-scripting the logic for selecting and utilizing these KBs. This conventional method, while effective, places limits on the flexibility and adaptability, and performance of the applications. But consider an alternative: applications that empower the LLM itself to determine which actions are necessary to generate a response. This approach not only harnesses the LLM's generative capabilities but also its ability to make contextual decisions on-the-fly, thereby opening up a more dynamic and responsive way to handle knowledge search tasks. As a result, you can build targeted KB's and use the power of LLMs to write application-specific queries that retrieve better, more relevant results. To demonstrate this approach, we'll build backed by a custom tool.
The chatbot will have access to a KB containing product information - any time you ask a question the chatbot determines could be answered using this tool it will query the KB and show both the in-progress query and retrieved results as custom UI elements before responding with the retrieved products as context. a simple chatbot We'll use Vercel's to render the chat window and Dewy to implement the information storage and retrieval used by our tool and as the knowledge store backing the tool. Generative UI library Dewy Why Dewy and Vercel? is designed to simplify the creation of custom LLM "tools" and to render rich, interactive UI elements that are specific to each tool the LLM chooses to use. Vercel's Generative UI library) What this means in practice is that your UI can dynamically adapt to the specific tools chosen by the LLM, displaying tool-specific information and progress indicators to the user. is an OSS knowledge base with the simplicity and ease-of-use of a document store: insert your documents and Dewy takes care of preparing them for semantic search.
Dewy's flexibility and ease of use allow developers to focus on building amazing user experiences rather than complex data-processing pipelines. Dewy Prerequisites Before diving into the tutorial, ensure you have the following prerequisites covered: Basic knowledge of Typescript and React A NextJS environment setup on your local machine A copy of Dewy running on your local machine (see Dewy's if you need help here). installation instructions Access to the platform. OpenAI API Set up your project : Create a NextJS app by running the following command in your terminal: Initialize a new NextJS project pnpm dlx create-next-app@canary rag-tool Navigate into your new project directory: cd rag-tool : Next, we'll install , Vercel's AI library, , OpenAI's official JavaScript SDK compatible with the Vercel Edge Runtime, and , the Dewy client library. Zod will be used to describe the input types for our custom tool. Install required packages ai openai dewy-ts pnpm install ai openai zod dewy-ts : Configure your OpenAI key and Dewy endpoint. Prepare your environment touch .env.local OPENAI_API_KEY=xxxxxxxxx
DEWY_ENDPOINT=localhost:8000 Create a custom tool First, create the custom tool the LLM will use to answer product-related questions.
This tool describes its purpose and parameters so the LLM knows when and how to use it, and defines the implementation and UI logic to complete the tool invocation. In this case, the tool will "get information about products", given a search query and a result count. When the LLM determines this tool should be used, it will generate the query string and choose how many results are appropriate, then execute the function defined below. render On execution, the tool searches Dewy for relevant products, then calls the LLM again to pick up where it left off. Since this process can take awhile, the function 's components indicating that the tool is being used and its progress. These components allow the UI to reflect the specific information relevant to this tool's outcomes. render yield import { z } from 'zod';
import { Dewy } from 'dewy-ts';
import { OpenAIStream } from 'ai'
import { Tokens } from 'ai/react';

import AssistantMessage from './AssistantMessage';
import ResultCard from './ResultCard';
import SearchCard from './SearchCard';

// Create a Dewy client.
const dewy = new Dewy({
    BASE: process.env.DEWY_ENDPOINT
})

// Implement the tool's logic.
// In this case, we search for the given query
// and return the `count` most similar chunks in the KB.
// The returned chunks are used in the `render` method below.
async function searchProducts(query: string, count: number) {
    const context = await dewy.kb.retrieveChunks({
        collection: "product_info",
        query: query,
        n: count,
    })
    return context.text_results.map(c => c.text)
}

// Define the behavior of the product search tool.
export default function productSearch(aiState, openai) {
    return {
        // A description of the tool.
        // This used by the LLM to decide when to use the tool
        description: 'Get information about products',

        // Parameters control how the tool behaves.
        // These values will be picked by the LLM,
        // so be sure to clearly explain what they're
        // used for.
        parameters: z.object({
            query: z.string().describe(```
                A description of a product 
                or what the product can be used for.
            ```),
            count: z.number().describe(```
                The number of products to return.
            ```),
        }).required(),

        // Configure the tool's behavior.
        // This function will be called after the LLM has
        // chosen to use the tool and generated values for
        // the parameters we configured above.
        render: async function* ({ query, count }) {
            // Let the user know we're looking for 
            // products related to their message
            yield <SearchCard query={query} count={count} />

            // Search for products related to the user's question
            const products = await searchProducts(query, count)

            // Update the message history 
            // with the results we found
            aiState.update([
                ...aiState.get(),
                {
                    role: "function",
                    name: "product_search",
                    content: JSON.stringify(products),
                }
            ]);

            // Now reply to the user.
            // The products we retrieved are part of the state 
            // provided as the messages parameter
            const resp = await openai.chat.completions.create({
                model: 'gpt-4-0125-preview',
                messages: aiState.get(),
                stream: true,
            })

            // Stream the results back as they're generated
            const stream = OpenAIStream(resp, {
                onFinal: (completion) => {
                    // Update the conversation history 
                    // once the full response is received
                    aiState.done([
                        ...aiState.get(),
                        {
                            role: "assistant",
                            content: completion,
                        }
                    ])
                }
            });

            // Display the response alongside 
            // the products provided to the model.
            return <div className="flex flex-col gap-2">
                <ResultCard query={query} results={context} />
                <AssistantMessage>
                    <Tokens stream={stream} />
                </AssistantMessage>
            </div>
        }
    }
} Setup the server-side message handler Configure a message handling action to receive new messages from the client.
This handler configures the initial state of the chatbot and defines the server's behavior when a message is received from the user. import { OpenAI } from 'openai';
import { createAI, createStreamableUI, getMutableAIState, render } from 'ai/rsc';

import productSearch from './productSearch';
import AssistantMessage from './AssistantMessage';

// Configure the LLM, in this case OpenAI
const openai = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY,
});
 
async function submitUserMessage(userInput: string) {
  'use server';
 
  // The AI state. This contains our message history,
  // and will be provided to the LLM each time a 
  // chat completion is generated.
  const aiState = getMutableAIState<typeof AI>();
  aiState.update([
    ...aiState.get(),
    {
      role: 'user',
      content: userInput,
    },
  ]);
 
  // Helper function for building streamable UI components.
  const ui = render({
    model: 'gpt-4-0125-preview',
    provider: openai,
    messages: [
      { role: 'system', content: `
        You are a product recommendation engine. 
        Respond only with information about products 
        retrieved using the "product_search" function role.
        `},
      { role: 'user', content: userInput }
    ],

    // This determines how generated text 
    // (as opposed to function calls) will
    // be rendered.
    text: ({ content, done }) => {
      if (done) {
        aiState.done([
          ...aiState.get(),
          {
            role: "assistant",
            content
          }
        ]);
      }
      return <AssistantMessage>{content}</AssistantMessage>
    },

    // Configure the tools available to the LLM
    tools: {
      product_search: productSearch(aiState, openai),
    }
  })
 
  return {
    id: Date.now(),
    display: ui
  };
}
 
// Define the initial state of the AI. It can be any JSON object.
const initialAIState: {
  role: 'user' | 'assistant' | 'system' | 'function';
  content: string;
  id?: string;
  name?: string;
}[] = [];
 
// The initial UI state that the client will keep track of, which contains the message IDs and their UI nodes.
const initialUIState: {
  id: number;
  display: React.ReactNode;
}[] = [{id: 0, display: <AssistantMessage>Hi! How can I help you today?</AssistantMessage>}];
 
// AI is a provider you wrap your application with so you can access AI and UI state in your components.
export const AI = createAI({
  actions: {
    submitUserMessage
  },
  // Each state can be any shape of object, but for chat applications
  // it makes sense to have an array of messages. Or you may prefer something like { id: number, messages: Message[] }
  initialUIState,
  initialAIState
}); Setup the chat UI on the client Finally, set up the chat UI on the client. Map over the UI state's messages and insert their property. display Configure a form to update the AI state, then call the server side handler. submitUserMessage 'use client'
 
import { useState } from 'react';
import { useUIState, useActions } from 'ai/rsc';
import type { AI } from './action';
import { Input } from '@/components/ui/input';
import { Button } from '@/components/ui/button';
 
import UserMessage from './UserMessage';
import { PlaneIcon } from './icons';

export default function Page() {
  const [inputValue, setInputValue] = useState('');
  const [messages, setMessages] = useUIState<typeof AI>();
  const { submitUserMessage } = useActions<typeof AI>();
 
  return (
    <>
      <div className="flex-1 overflow-auto px-4 mt-4">
        <div className="grid gap-4 md:gap-8">
          {
            messages.map((message) => (
              <div key={message.id}>
                {message.display}
              </div>
            ))
          }

        </div>
      </div>

      <form onSubmit={async (e) => {
        e.preventDefault();

        // Add user message to UI state
        setMessages((currentMessages) => [
          ...currentMessages,
          {
            id: Date.now(),
            display: <UserMessage>{inputValue}</UserMessage>,
          },
        ]);

        // Submit and get response message
        const responseMessage = await submitUserMessage(inputValue);
        setMessages((currentMessages) => [
          ...currentMessages,
          responseMessage,
        ]);

        setInputValue('');
      }}>
        <div className="border-t-2">
          <div className="flex items-center h-14 px-4">
            <Input
              className="rounded-full flex-1 min-w-0 bg-gray-200 dark:bg-gray-800"
              placeholder="Type a message..."
              value={inputValue}
              type="text"
              onChange={(event) => {
                setInputValue(event.target.value)
              }}
            />
            <Button className="ml-2 h-8 w-8 rounded-full bg-gray-200 dark:bg-gray-800" size="icon">
              <PlaneIcon className="h-4 w-4" />
              <span className="sr-only">Send message</span>
            </Button>
          </div>
        </div>
      </form>
    </>
  )
} To keep this post from going any longer than it is already, several basic UI components have been omitted, but you can checkout the full implementation in the of the Dewy repo. examples directory Conclusion In conclusion, this blog post has guided you through the process of building a Retrieval-Augmented Generation (RAG) tool using Vercel's Generative UI components. This approach moves away from the traditional methods of pre-scripting logic for selecting and utilizing knowledge bases. By empowering the LLM to determine which knowledge bases are necessary and how best to use them, we've showcased a more dynamic, efficient, and user-centric way of handling information retrieval tasks. By utilizing Dewy as our knowledge base, we've emphasized the importance of a flexible, easy-to-use platform for storing, organizing, and retrieving information. This synergy between Dewy's streamlined data management and Vercel's dynamic UI rendering paves the way for developers to create more intelligent, responsive, and user-friendly applications. Also appears . here

How to Build a "RAG Tool" With Vercel's Generative UI Components

When to Use the Logger Class? : Detailed and customizable logs simplify identifying and rectifying issues in complex applications. Debugging and Troubleshooting : Logging is essential for monitoring behavior, identifying performance bottlenecks, and analyzing usage patterns in production environments. Monitoring and Analysis : Comprehensive logging is vital for auditing and security analysis in applications handling sensitive information or needing to adhere to specific compliance requirements. Security and Compliance How to Use the Logger Class? Save the following code in a file named , and then open another file in the same root directory and use to import the package. Logger.py Python main.py from Logger import Logger Logger.py import logging
from logging import handlers

class Logger(object):
    level_relations = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'crit': logging.CRITICAL
    }  # relationship mapping

    def __init__(self, filename, level='info', when='D', backCount=3,
                 fmt='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s'):
        self.logger = logging.getLogger(filename)
        format_str = logging.Formatter(fmt)  # Setting the log format
        self.logger.setLevel(self.level_relations.get(level))  # Setting the log level
        console_handler = logging.StreamHandler()  # on-screen output
        console_handler .setFormatter(format_str)  # Setting the format
        th = handlers.TimedRotatingFileHandler(filename=filename, when=when, backupCount=backCount,encoding='utf-8')  # automatically generates the file at specified intervals
        th.setFormatter(format_str)  # Setting the format
        self.logger.addHandler(sh)  # Add the object to the logger
        self.logger.addHandler(th) This code defines a custom Logger class that can be used to create and configure loggers in Python. The class is initialized with several parameters: Logger : The name of the log file where the logs will be saved. filename : The logging level, which determines the severity of the messages to be logged (e.g., 'debug', 'info', 'warning', 'error', 'crit'). level : The interval at which the log files should be rotated (e.g., 'S' for seconds, 'M' for minutes, 'H' for hours, 'D' for days, 'W' for weeks). when : The number of backup log files to retain. backCount : The format of the log messages, which includes the time, file path, line number, log level, and the log message itself. fmt Inside the method, the logger is configured as follows: __init__ The logging level is set based on the provided . level A format string is created for formatting the log messages. A console handler is created for logging messages to the console. A rotating file handler is created for logging messages to the specified log file. Finally, the logger is set up to use both the console handler and the rotating file handler for logging messages. main.py from Logger import Logger

log = Logger('log\\all.log', level='debug')
log_error = Logger('log\\all.log', level='debug')

log.logger.info("This is an info message)")
log_error.info("This is an error message") This Python code shows how to use the Logger package to write log information to a file. It first initializes two Logger objects, one for general logging and one for error logging. It then uses these objects to write log messages at different levels, such as info and error. The log messages are written to the specified log file, allowing for easy monitoring and troubleshooting of the application. Customizing Parameters The Logger class provides several parameters that can be customized based on the specific logging needs. These parameters include: : The name of the log file where the logs will be written. filename : The log level, which determines the severity of the logs (e.g., debug, info, warning, error, crit). level : The frequency at which the log files should be rotated (e.g., 'D' for daily, 'H' for hourly). when : The number of backup log files to retain after rotation. backCount : The format of the log messages, allowing customization of the timestamp, log level, file name, line number, and message content. fmt By adjusting these parameters, the Logger class can be tailored to meet the specific logging requirements of different projects and applications. Example 1: Customizing Log Level and Format log = Logger('custom.log', level='warning', fmt='%(asctime)s - %(levelname)s - %(message)s')
log.logger.debug("This is a debug message - won't be logged")
log.logger.warning("This is a warning message - will be logged") Example 2: Specifying Log Rotation Parameters log = Logger('rotation.log', when='midnight', backCount=7)
log.logger.info("Logging with rotation based on time") Conclusion Logging is crucial in software development for monitoring, troubleshooting, analyzing, and securing applications. Python's logging class offers a convenient and customizable way to implement logging with various configuration options. This class enables developers to tailor logging setups to meet project needs, ensuring comprehensive and efficient logging for debugging, monitoring, and compliance purposes. The Logger class empowers developers to harness the power of logging for robust application development and maintenance. Happy logging! Thank you for taking the time to explore data-related insights with me. I appreciate your engagement. If you find this information helpful, I invite you to follow me or connect with me on . Happy exploring!👋 LinkedIn Also published . here

Python Logging: Debugging, Monitoring, and Security with the Logger Class

Recently, we took over a data integration project where the upstream data was delivered to Kafka. Initially, we chose the SpringBoot+Flink approach to process the incoming data (referred to as Solution 1 below). However, after facing efficiency issues in the production environment, we turned our attention to the open-source project SeaTunnel. We discovered that SeaTunnel supports Kafka as a source and started exploring and testing it. In our development environment, we experimented with over 5 million records and achieved an efficiency of around 10,000 records per second. Consequently, we abandoned Solution 1 and adopted SeaTunnel for integrated data processing (referred to as Solution 2 below). During our exploration of SeaTunnel, we identified two methods. Method 2, compared to Method 1, offers a more comprehensive solution that handles various scenarios without concerns about data misalignment caused by unexpected characters within field values. Version Information: SeaTunnel: apache-seatunnel-2.3.2-SNAPSHOT Engines: Flink 1.16.2 Zeta: Consistent with SeaTunnel Comparison Based on Solution 2, two derived methods were developed: After extensive exploration and validation in our production environment, we recommend using Method 2 of Solution 2. Now, let's dive into the main content and explore the two methods in Solution 2, allowing you to experience the magic of SeaTunnel. Solution 1: Kafka Complex JSON Parsing with Spring Boot + Flink There are many online examples available for this approach, so we won't go into much detail here. Solution 2: Kafka Complex JSON Parsing with SeaTunnel Before we proceed, let's take a look at a sample JSON data delivered to Kafka by the upstream party (part of the data has been redacted): "type": "D***",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
    "data": {
        "LSH": "187eb13****l0214723",
        "NSRSBH": "9134****XERG56",
        "QMYC": "01*****135468",
        "QMZ": "1Zr*****UYGy%2F5bOWtrh",
        "QM_SJ": "2023-05-05 16:42:10.000000",
        "YX_BZ": "Y",
        "ZGHQ_BZ": "Y",
        "ZGHQ_SJ": "2023-06-26 16:57:17.000000",
        "SKSSQ": 202304,
        "SWJG_DM": "",
        "SWRY_DM": "00",
        "CZSJ": "2023-05-05 16:42:10.000000",
        "YNSRSBH": "9134****XERG56",
        "SJTBSJ": "2023-06-26 19:29:59.0,00",
        "SJCZBS": "I"
    },
    "beforeData": null,
    "headers": {
        "operation": "INSERT",
        "changeSequence": "12440977",
        "timestamp": "2023-06-26T19:29:59.673000",
        "streamPosition": "00****3.16",
        "transactionId": "000***0006B0002",
        "changeMask": "0FFF***FF",
        "columnMask": "0F***FFFF",
        "transactionEventCounter": 1,
        "transactionLastEvent": false
    }
} Method 1: Parsing without UDF Functions Issue: Field values may contain delimiters, such as ','. This can lead to data misalignment during storage. This method primarily utilizes various transformation plugins available in the official transform-v2. It includes the , , and implementations. Replace Split SQL SeaTunnel Script: (ybjc_qrqm.conf) job.mode = "STREAMING"
job.name = "kafka2mysql_ybjc"
execution.checkpoint.interval = 60000

source {
  Kafka {
    result_table_name = "DZFP_***_QRQM1"
    topic = "DZFP_***_QRQM"
    bootstrap.servers = "centos1:19092,centos2:19092,centos3:19092"
    schema = {
      fields {
        message =  {
            data = {
                LSH = "string",
                NSRSBH =  "string",
                QMYC =  "string",
                QMZ =  "string",
                QM_SJ =  "string",
                YX_BZ =  "string",
                ZGHQ_BZ =  "string",
                ZGHQ_SJ =  "string",
                SKSSQ =  "string",
                SWJG_DM = "string",
                SWRY_DM = "string",
                CZSJ = "string",
                YNSRSBH = "string",
                SJTBSJ = "string",
                SJCZBS = "string"
            }
        }
      }  
    }
	  start_mode = "earliest"
    kafka.config = {
      auto.offset.reset = "earliest"
      enable.auto.commit = "true"
      max.partition.fetch.bytes = "5242880"
      session.timeout.ms = "30000"
      max.poll.records = "100000"
    }
  }
}

transform {
 Replace {
    source_table_name = "DZFP_***_QRQM1"
    result_table_name = "DZFP_***_QRQM2"
    replace_field = "message"
    pattern = "[["
    replacement = ""
  }
  Replace {
    source_table_name = "DZFP_***_QRQM2"
    result_table_name = "DZFP_***_QRQM3"
    replace_field = "message"
    pattern = "]]"
    replacement = ""
  }
  
  Split {
    source_table_name = "DZFP_***_QRQM3"
    result_table_name = "DZFP_***_QRQM4"
    separator = ","
    split_field = "message"
    output_fields = [zwf1,zwf2,zwf3,zwf4,zwf5,nsrsbh,qmyc,qmz,qm_sj,yx_bz,zghq_bz,zghq_sj,skssq,swjg_dm ,swry_dm ,czsj ,ynsrsbh ,sjtbsj ,sjczbs]
  }
  
  sql{
   source_table_name = "DZFP_***_QRQM4"
   query = "select replace(zwf5 ,'fields=[','') as lsh,nsrsbh,trim(qmyc) as qmyc,qmz,qm_sj,yx_bz, zghq_bz,zghq_sj,skssq,swjg_dm ,swry_dm ,czsj ,ynsrsbh ,sjtbsj ,replace(sjczbs,']}]}','') as sjczbs  from DZFP_DZDZ_QRPT_YWRZ_QRQM4 where skssq <> ' null'"
   result_table_name = "DZFP_***_QRQM5"
  }
  
}

sink {
	Console {
		source_table_name = "DZFP_***_QRQM5"
	}
	
    jdbc {
	    source_table_name = "DZFP_***_QRQM5"
        url = "jdbc:mysql://localhost:3306/dbname?serverTimezone=GMT%2b8"
        driver = "com.mysql.cj.jdbc.Driver"
        user = "user"
        password = "pwd"
        batch_size = 200000
        database = "dbname"
        table = "tablename"
        generate_sink_sql = true
        primary_keys = ["nsrsbh","skssq"]
    } 
} Data can be successfully written: Kafka Source Data: Target Data in TiDB: However, when we simulate sending data to Kafka, specifically setting a ',' as part of the 'SJTBSJ' field: Original value: Modified value: 2023-06-26 19:29:59.000 2023-06-26 19:29:59.0,00 Command to produce data to the topic: kafka-console-producer.sh --topic DZFP_***_QRQM --broker-list centos1:19092,centos2:19092,centos3:19092 Data sent: "type": "D***",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
    "data": {
        "LSH": "187eb13****l0214723",
        "NSRSBH": "9134****XERG56",
        "QMYC": "01*****135468",
        "QMZ": "1Zr*****UYGy%2F5bOWtrh",
        "QM_SJ": "2023-05-05 16:42:10.000000",
        "YX_BZ": "Y",
        "ZGHQ_BZ": "Y",
        "ZGHQ_SJ": "2023-06-26 16:57:17.000000",
        "SKSSQ": 202304,
        "SWJG_DM": "",
        "SWRY_DM": "00",
        "CZSJ": "2023-05-05 16:42:10.000000",
        "YNSRSBH": "9134****XERG56",
        "SJTBSJ": "2023-06-26 19:29:59.0,00",
        "SJCZBS": "I"
    },
    "beforeData": null,
    "headers": {
        "operation": "INSERT",
        "changeSequence": "12440977",
        "timestamp": "2023-06-26T19:29:59.673000",
        "streamPosition": "00****3.16",
        "transactionId": "000***0006B0002",
        "changeMask": "0FFF***FF",
        "columnMask": "0F***FFFF",
        "transactionEventCounter": 1,
        "transactionLastEvent": false
    }
} After writing the data, we notice that the data is misaligned: Conclusion: This issue can still occur in production environments when fields contain delimiters like commas in addresses or remarks. Therefore, this approach is not recommended as it poses significant risks to the data. It can be used for handling simple data as an alternative approach. Method 2: Parsing with UDF Functions This method extends SeaTunnel's capabilities using UDF functions ( ), enabling nested JSON source data parsing from Kafka. It simplifies the ST script configuration significantly. https://seatunnel.apache.org/docs/2.3.2/transform-v2/sql-udf ST Script: (ybjc_qrqm_yh.conf) execution.parallelism = 5
job.mode = "STREAMING"
job.name = "kafka2mysql_ybjc_yh"
execution.checkpoint.interval = 60000
}

source {
    Kafka {
        result_table_name = "DZFP_***_QRQM1"
        topic = "DZFP_***_QRQM"
        bootstrap.servers = "centos1:19092,centos2:19092,centos3:19092"
        schema = {
        fields {
             message =  {
                data = "map<string,string>"
             }
        }  
        }
        start_mode = "earliest"
        kafka.config = {
        auto.offset.reset = "earliest"
        enable.auto.commit = "true"
        max.partition.fetch.bytes = "5242880"
        session.timeout.ms = "30000"
        max.poll.records = "100000"
        }
    }
}
transform {
    sql{
        source_table_name = "DZFP_***_QRQM1"
        result_table_name = "DZFP_***_QRQM2"
        query = "select qdmx(message,'lsh') as lsh,qdmx(message,'nsrsbh') as nsrsbh,qdmx(message,'qmyc') as qmyc,qdmx(message,'qmz') as qmz,qdmx(message,'qm_sj') as qm_sj,qdmx(message,'yx_bz') as yx_bz,qdmx(message,'zghq_bz') as zghq_bz,qdmx(message,'zghq_sj') as zghq_sj,qdmx(message,'skssq') as skssq,qdmx(message,'swjg_dm') as swjg_dm,qdmx(message,'swry_dm') as swry_dm,qdmx(message,'czsj') as czsj,qdmx(message,'ynsrsbh') as ynsrsbh, qdmx(message,'sjtbsj') as sjtbsj,qdmx(message,'sjczbs') as sjczbs  from  DZFP_DZDZ_QRPT_YWRZ_QRQM1"
    }
}

    
sink {
    Console {
        source_table_name = "DZFP_***_QRQM2"
    }
    jdbc {
        source_table_name = "DZFP_***_QRQM2"
        url = "jdbc:mysql://localhost:3306/dbname?serverTimezone=GMT%2b8"
        driver = "com.mysql.cj.jdbc.Driver"
        user = "user"
        password = "pwd"
        batch_size = 200000
        database = "dbname"
        table = "tablename"
        generate_sink_sql = true
        primary_keys = ["nsrsbh","skssq"]
    } 
} Executing the script and checking the results, we can see that the data is not misaligned and remains in the original field ('sjtbsj'): The specific UDF function is implemented as follows. Maven dependencies are as follows: <dependency>
    <groupId>org.apache.seatunnel</groupId>
    <artifactId>seatunnel-transforms-v2</artifactId>
    <version>2.3.2</version>
    <scope>provided</scope>
  </dependency>

  <dependency>
    <groupId>org.apache.seatunnel</groupId>
    <artifactId>seatunnel-api</artifactId>
    <version>2.3.2</version>
  </dependency>

  <dependency>
    <groupId>cn.hutool</groupId>
    <artifactId>hutool-all</artifactId>
    <version>5.8.20</version>
  </dependency>

  <dependency>
    <groupId>com.google.auto.service</groupId>
    <artifactId>auto-service-annotations</artifactId>
    <version>1.1.1</version>
    <optional>true</optional>
    <scope>compile</scope>
  </dependency>

  <dependency>
    <groupId>com.google.auto.service</groupId>
    <artifactId>auto-service</artifactId>
    <version>1.1.1</version>
    <optional>true</optional>
    <scope>compile</scope>
  </dependency>
</dependencies> UDF-specific implementation in Java: import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.google.auto.service.AutoService;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.transform.sql.zeta.ZetaUDF;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@AutoService(ZetaUDF.class)
public class QdmxUDF implements ZetaUDF {

    @Override
    public String functionName() {
        return "QDMX";
    }

    @Override
    public SeaTunnelDataType<?> resultType(List<SeaTunnelDataType<?>> list) {
        return BasicType.STRING_TYPE;
    }

    // list Example of parameters: (which is the data parsed by Kafka)
    //SeaTunnelRow{tableId=, kind=+I, fields=[{key1=value1,key2=value2,.....}]}
    @Override
    public Object evaluate(List<Object> list) {
        String str = list.get(0).toString();
        //1 Remove the prefix
        str = StrUtil.replace(str, "SeaTunnelRow{tableId=, kind=+I, fields=[{", "");
        //2 Remove the suffix
        str = StrUtil.sub(str, -3, 0);
        // 3 build Map key value
        Map<String, String> map = parseToMap(str);
        if ("null".equals(map.get(list.get(1).toString())))
            return "";
        // 4 return the value of the key
        return map.get(list.get(1).toString());
    }

    public static Map<String, String> parseToMap(String input) {
        Map<String, String> map = new HashMap<>();
        // Remove the curly brackets and remove them during the string stage.
        // input = input.replaceAll("[{}]", "");
        // Split key-value pairs
        String[] pairs = input.split(", ");

        for (String pair : pairs) {
            String[] keyValue = pair.split("=");
            if (keyValue.length == 2) {
                String key = keyValue[0].trim().toLowerCase();
                String value = keyValue[1].trim();
                map.put(key, value);
            }
        }
        return map;
    }
} Then, for packaging, use the following command: mvn -T 8 clean install -DskipTests -Dcheckstyle.skip -Dmaven.javadoc.skip=true Check the directory to see if the @AutoService annotation generates the corresponding SPI interface. META-INF/services If it is as shown below, then the packaging is successful! If not, the packaging fails, and the UDF function cannot be used. You can refer to my packaging plugin: <plugins>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-site-plugin</artifactId>
      <version>3.7</version>
      <dependencies>
        <dependency>
          <groupId>org.apache.maven.doxia</groupId>
          <artifactId>doxia-site-renderer</artifactId>
          <version>1.8</version>
        </dependency>
      </dependencies>
    </plugin>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-compiler-plugin</artifactId>
      <version>3.8.1</version>
      <configuration>
        <annotationProcessorPaths>
          <path>
            <groupId>com.google.auto.service</groupId>
            <artifactId>auto-service</artifactId>
            <version>1.1.1</version>
          </path>
        </annotationProcessorPaths>
      </configuration>
    </plugin>
  </plugins>
</build> Finally, place the packaged JAR file in the $ directory. Since my UDF function depends on a third-party JAR, it should be uploaded together. If using the Zeta cluster, you need to restart the Zeta cluster for it to take effect. For other engines, it will take effect in real time. {SEATUNNEL_HOME}/lib The successful upload will look like this: Note: The "hutool-all" JAR file can be included in the project. I uploaded two for convenience. java_study In conclusion, it is recommended to to parse the nested JSON data from Kafka sources. use the extension of UDF functions Also published . here

Resolving Complex JSON in Kafka Source Using Apache SeaTunnel

I've written a few posts lately about multi-host live streams with Amazon Interactive Video Service (Amazon IVS). It's an exciting feature that opens up worlds of possibilities that simply weren't available until recently. We first looked at how to . create a multi-host live chat application Next, we saw how to broadcast that live chat session to an Amazon IVS channel. When we looked at adding chat participants to the broadcast client in that last post, you probably noticed that I cheated a bit and hardcoded the values that tell the broadcast client the size and position of the participant's video on the client. VideoComposition Well - cheated is a strong word - let's say that I intentionally simplified the code to focus on the process of broadcasting a live chat session. Essentially what we're looking for here is modifying the size and position of the participant's video in the broadcast so that when there is one video, the layout will look something like this: But when there are two videos, the layout will change to something like this: And when there are five: You get the idea - a dynamic layout that changes based on the amount of participants. In this post, we'll take a look at one approach you could utilize to make creating a dynamic layout a bit easier. We'll build off of the solution in the last post, so if you haven't read that post yet, it's probably a good idea to do that now. In the last post, we listened for an event called . In the event handler for that event, we added our participants to the DOM and rendered the audio and video to the instance. STAGE_PARTICIPANT_STREAMS_ADDED IVSBroadcastClient In order to render a dynamic layout, we'll need to track how many participants are currently in the session, so we'll add an array called as a global variable. Let's modify the event handler to push the current participant id to that array. participantIds stage.on(StageEvents.STAGE_PARTICIPANT_STREAMS_ADDED, (participant, streams) => {

  //add participant id to array
  participantIds.push(participant.id);

  renderParticipant(participant, streams);
  renderVideosToClient(participant, streams.find(s => s.streamType === StreamType.VIDEO));
  renderAudioToClient(participant, streams.find(s => s.streamType === StreamType.AUDIO));
  updateVideoCompositions();
}); In the last post, I mentioned that the method was not shown because the implementation would vary. We'll talk about one possible implementation in just a bit. updateVideoCompositions() For now, let's take a look at how we can get a dynamic layout config instead of hardcoding it as we did in the last post. One way to obtain a dynamic size and position is to loop over the participant array and calculate them based on the number of participants, the size of the , and the desired amount of rows, columns, and padding. But, ? <canvas> why That sounds like a lot of difficult code and unnecessary work when you realize that these values never change. If you have one participant, the video will be a fixed size and centered in the . <canvas> It doesn't matter how many participants get added - the layout for each video will always be the same for a given number of participants. So why waste time and CPU cycles when we could pre-calculate these values and store them in an array of arrays? For my demo, I spent some time determining the best values with an intensive 30 minutes with a pen, paper, and calculator to determine the composition values for each possible layout. Please note: I was a maths or art major as evidenced by the following sketch. not For this demo, I capped my live stream at only showing videos for the first 6 participants. Your use case may dictate something different, but having more than 6 participant videos in one live stream becomes a bit too busy in my experience. Here is the result of my calculations: const layouts = [
  [{ height: 720, width: 1280, x: 320, y: 180 }],
  [{ height: 450, width: 800, x: 80, y: 315 }, { height: 450, width: 800, x: 1040, y: 315 }],
  [{ height: 450, width: 800, x: 80, y: 45 }, { height: 450, width: 800, x: 1040, y: 45 }, { height: 450, width: 800, x: 560, y: 585 }],
  [{ height: 450, width: 800, x: 80, y: 45 }, { height: 450, width: 800, x: 1040, y: 45 }, { height: 450, width: 800, x: 80, y: 585 }, { height: 450, width: 800, x: 1040, y: 585 }],
  [{ height: 337, width: 600, x: 20, y: 100 }, { height: 337, width: 600, x: 650, y: 100 }, { height: 337, width: 600, x: 1280, y: 100 }, { height: 337, width: 600, x: 340, y: 640 }, { height: 337, width: 600, x: 980, y: 640 }],
  [{ height: 337, width: 600, x: 20, y: 100 }, { height: 337, width: 600, x: 650, y: 100 }, { height: 337, width: 600, x: 1280, y: 100 }, { height: 337, width: 600, x: 20, y: 640 }, { height: 337, width: 600, x: 650, y: 640 }, { height: 337, width: 600, x: 1280, y: 640 }]
]; That might look overwhelming, but consider that each element in the outer array element contains an array of compositions for each video. If there are 3 participants, we can reference the third element in the outer array, and the position of the participant id in the array will determine which composition will apply to that video. participantIds We can modify our function to grab the proper composition and use those values when we add the video to the broadcast client. renderVideosToClient() const renderVideosToClient = async (participant, stream) => {
  const participantId = participant.id;
  const videoId = `video-${participantId}`;

  // get the index of this participantId
  const pIdx = participantIds.indexOf(participantId);

  let composition = layouts[participantIds.length - 1][pIdx];
  config.index = 2;

  const mediaStream = new MediaStream();
  mediaStream.addTrack(stream.mediaStreamTrack);
  broadcastClient.addVideoInputDevice(mediaStream, videoId, composition);
}; But remember - if we only do this when a participant is added, the previous video compositions will still reflect the composition that was applied when they were added. That is where the function comes into the picture. updateVideoCompositions() Here we loop over the array, grab the proper composition from , and use the ( ) method of the . participantIds layouts updateVideoDeviceComposition() docs broadcastClient const updateVideoCompositions = async () => {
  let idx = 0;
  for (const p of participantIds) {
    const videoId = `video-${p}`;
    let config = layouts[filteredParticipantIds.length - 1][idx];
    config.index = 2;
    broadcastClient.updateVideoDeviceComposition(videoId, config);
    idx = idx + 1;
  }
}; We should also make sure that when a participant leaves the stage that we remove the participant id from the array and again update the composition for all videos. stage.on(StageEvents.STAGE_PARTICIPANT_STREAMS_REMOVED, (participant, streams) => {
  const participantId = participant.id;

  // remove participant id from array
  const pIdx = participantIds.findIndex(id => id === participantId);
  participantIds.splice(pIdx, 1);

  const videoTrackId = `video-${participantId}`;
  const audioTrackId = `audio-${participantId}`;
  if (broadcastClient.getVideoInputDevice(videoTrackId)) broadcastClient.removeVideoInputDevice(videoTrackId);
  if (broadcastClient.getAudioInputDevice(audioTrackId)) broadcastClient.removeAudioInputDevice(audioTrackId);
  const videoId = `${participantId}-video`;
  document.getElementById(videoId).closest('.participant-col').remove();
  updateVideoCompositions(); 
}); As mentioned above, you'll most likely want to limit the amount of videos that are added to the live stream via the broadcast client. You might want to add a static image instead of the final video to show that there are more participants than what is shown: Summary In this post, we learned one approach for dynamic layouts when broadcasting a multi-host stage with Amazon IVS. In a future post, we'll look at additional options for broadcasting with multiple hosts. As always, if you have any questions or comments, please leave them below. Also published here

How to Create a Dynamic Layout for Multi Host Broadcasts With Amazon IVS

Cómo crear un diseño dinámico para transmisiones de varios hosts con Amazon IVS

Comment créer une mise en page dynamique pour les diffusions multi-hôtes avec Amazon IVS

अमेज़ॅन आईवीएस के साथ मल्टी होस्ट ब्रॉडकास्ट के लिए डायनामिक लेआउट कैसे बनाएं

Amazon IVS を使用してマルチホスト ブロードキャスト用の動的レイアウトを作成する方法

Como criar um layout dinâmico para transmissões de vários hosts com o Amazon IVS

Cách tạo bố cục động cho truyền phát nhiều máy chủ với Amazon IVS

如何使用 Amazon IVS 创建多主机广播的动态布局

In our we learned how to create a virtual "stage" to create a real-time video chat experience for up to 12 participants with Amazon Interactive Video Service (Amazon IVS). As a standalone feature, that's pretty powerful and enables us to add real-time collaboration to our applications. However, this feature was created to empower developers to easily create collaborative live streaming experiences similar to the "Guest Star" feature on Twitch. In this post, we'll build on the previous demo to combine the audio and video feeds from all participants into a single feed and broadcast that feed to an Amazon IVS channel. last post If you haven't yet read the previous post, you should do that before moving forward with this post. To recap, in that post we learned how to: Create a stage resource with the AWS SDK for JavaScript (v3) Create a stage participant tokens with the AWS SDK for JavaScript (v3) Use the Web Broadcast SDK to connect to the virtual stage for real-time video chat between participants The next step to creating a collaborative live streaming experience is to combine (or "composite") both the local and remote participants into a single stream that can be published to an Amazon IVS channel. For this we can also use the Web Broadcast SDK, so let's see how it's done. Creating a Broadcast Client If you recall, in the last post we had several functions called inside of a handler that enabled permissions, obtained devices, configured the instance, and handled joining the stage. We'll add one more method to this flow called which we can use to create an instance of the . We'll need a element in our markup for the combined stream so that our participants can preview what will ultimately be broadcast to the Amazon IVS channel. DOMContentLoaded Stage initBroadcastClient() IVSBroadcastClient <canvas> const initBroadcastClient = async () => {
  broadcastClient = IVSBroadcastClient.create({
    streamConfig: IVSBroadcastClient.STANDARD_LANDSCAPE,
    ingestEndpoint: '[YOUR INGEST ENDPOINT]',
  });

  const previewEl = document.getElementById('broadcast-preview');
  broadcastClient.attachPreview(previewEl);

  const bgImage = new Image();
  bgImage.src = '/images/stage_bg.png';
  broadcastClient.addImageSource(bgImage, 'bg-image', { index: 0 });
}; To make things a little more visually appealing, I've used to add a background image to the stream. The method receives three arguments: the image, a unique name for the source, and a object that is used to define the (or 'layer') for the source. If you check the for , you'll also note that it can contain values for the , , , and position for the source. We'll take advantage of those properties in just a bit when we add our video layers for each participant. addImageSource() addImageSource() VideoComposition index docs VideoComposition height width x y Adding Participant Audio and Video to the Broadcast Client Next, we're going to add the audio and video for each participant to the broadcast client. We'll do this inside of the handler that we defined in the previous post. Modify that function to add calls to two new functions. StageEvents.STAGE_PARTICIPANT_STREAMS_ADDED stage.on(StageEvents.STAGE_PARTICIPANT_STREAMS_ADDED, (participant, streams) => {
  renderParticipant(participant, streams);
  renderVideosToClient(participant, streams.find(s => s.streamType === StreamType.VIDEO));
  renderAudioToClient(participant, streams.find(s => s.streamType === StreamType.AUDIO));
}); Now let's create the function. Here, I'm hardcoding the to values appropriate for a single participant. In your application, you'll want to dynamically calculate the , , , and values depending on the amount of users currently participating in the conversation. renderVideosToClient() VideoComposition height width x y const renderVideosToClient = async (participant, stream) => {
  const participantId = participant.id;
  const videoId = `video-${participantId}`;
  const composition = { 
    index: 1,
    height: 984, 
    width: 1750, 
    x: 85, 
    y: 48 
  };
  const mediaStream = new MediaStream();
  mediaStream.addTrack(stream.mediaStreamTrack);
  broadcastClient.addVideoInputDevice(mediaStream, videoId, composition);
}; The function looks similar, but uses the method of the SDK to add the audio track. renderAudioToClient() addAudioInputDevice() const renderAudioToClient = async (participant, stream) => {
  const participantId = participant.id;
  const audioTrackId = `audio-${participantId}`;
  const mediaStream = new MediaStream();
  mediaStream.addTrack(stream.mediaStreamTrack);
  broadcastClient.addAudioInputDevice(mediaStream, audioTrackId)
}; At this point, the stage is ready to be broadcast to a channel by calling . We'll also need to handle removing participants from the when they leave a stage. For this, update the handler for . broadcastClient.startBroadcast('[YOUR STREAM KEY]') broadcastClient StageEvents.STAGE_PARTICIPANT_STREAMS_REMOVED stage.on(StageEvents.STAGE_PARTICIPANT_STREAMS_REMOVED, (participant, streams) => {
  const videoTrackId = `video-${participant.id}`;
  const audioTrackId = `audio-${participant.id}`;
  if (broadcastClient.getVideoInputDevice(videoTrackId)) broadcastClient.removeVideoInputDevice(videoTrackId);
  if (broadcastClient.getAudioInputDevice(audioTrackId)) broadcastClient.removeAudioInputDevice(audioTrackId);
  const videoId = `${participant.id}-video`
  document.getElementById(videoId).closest('.participant-col').remove();
  updateVideoCompositions(); // function not defined in demo, implementation will vary
}); Here's how an implementation might look with a single participant. Note that each stage participant would be shown on the bottom screen, and the composite view to be broadcast is shown above. And when multiple participants have joined the virtual stage, the application adjusts the layout to accommodate each participant. When the 'host' participant clicks the 'Broadcast' button, the combined conversation will be broadcast to the Amazon IVS channel as a composite view with all participants audio and video combined into a single stream. Summary In this post, we learned how to create a live stream broadcast with the audio and video from multiple remote participants. In a future post, we'll examine alternative options for creating the composite stream and broadcasting it to an Amazon IVS channel.

How to Live Stream With Multiple Hosts using a Browser and Amazon IVS

Cómo transmitir en vivo con múltiples hosts usando un navegador y Amazon IVS

Comment diffuser en direct avec plusieurs hôtes à l'aide d'un navigateur et d'Amazon IVS

एक ब्राउज़र और अमेज़ॅन आईवीएस का उपयोग करके एकाधिक होस्ट के साथ लाइव स्ट्रीम कैसे करें

ブラウザと Amazon IVS を使用して複数のホストでライブ ストリーミングする方法

Como transmitir ao vivo com vários hosts usando um navegador e o Amazon IVS

Cách phát trực tiếp với nhiều máy chủ bằng Trình duyệt và Amazon IVS

如何使用浏览器和 Amazon IVS 与多个主机进行直播

We recently support for collaborative live streams via a new virtual resource called a stage with Amazon Interactive Video Service (Amazon IVS). This feature opens up a multitude of possibilities that were not easy (or even possible) to implement in the past. announced This feature gives live streamers the ability to invite guests into their stream, collaborate with other content creators, or even create a "call-in" style show with audio-only participants. In this post, we'll look at how to get started creating a web application with multiple hosts with the Web Broadcast SDK. As always, the are a great starting point, but it always helps to walk through a demo application, so that's what we'll focus on in this post. docs For this demo, we'll build a web application that allows multiple participants to view and talk to one another. In a future post, we'll add the ability to broadcast the conversation to an Amazon IVS channel where others can view the conversation. At a high level, here are the steps that we'll take to create a virtual "stage" application for real-time collaboration between multiple hosts. Create a via the SDK stage Issue token(s) for stage participants Connect to the on the client stage Render participant's audio and video when they join a stage The Server Side We will utilize the new module ( ) of the AWS SDK for JavaScript (v3) to create the stage resource and participant tokens. @aws-sdk/client-ivs-realtime docs Here is an that you can use to handle these steps, but for my demo, I decided to simplify things a bit and created an Express application that can be used to host the API and my front-end. example of a serverless application Creating a Stage Resource via the AWS SDK for JavaScript (v3) In my Express app, I created a service called . This service caches existing and in memory to avoid the need for a database to persist them. IvsRealtimeService stages stageParticipants In production, you'd want to store these values in a persistent data store to avoid the volatility and limitations of caching the values in memory (again, refer to the to learn more about using Amazon DynamoDB to persist stages and tokens). serverless demo My app has one route called that can be used to a stage . This route uses my to check for an existing stage by that name, and either returns the existing stage or creates a new stage. /ivs-stage POST name IvsRealtimeService router.post('/ivs-stage', async function (req, res, next) {
  const body = req.body;
  const stageName = body.name;
  let stage = 
    ivsRealtimeService.stageExists(stageName) ? 
      ivsRealtimeService.getStage(stageName) : 
      await ivsRealtimeService.createStage(stageName);
  res.json(stage);
}); The and methods look like so: stageExists() getStage() import { CreateParticipantTokenCommand, CreateStageCommand, IVSRealTimeClient } from "@aws-sdk/client-ivs-realtime";

const config = {
  credentials: {
    accessKeyId: process.env.ACCESS_KEY,
    secretAccessKey: process.env.SECRET_KEY,
  }
};

export default class IvsRealtimeService {
  ivsRealtimeClient = new IVSRealTimeClient(config);
  stages = {};
  stageParticipants = {};

  getStage(name) {
    return this.stages[name];
  }

  stageExists(name) {
    return this.stages.hasOwnProperty(name);
  }
} And the method sends a via the . createStage() CreateStageCommand ivsRealtimeClient async createStage(name) {
  const createStageRequest = new CreateStageCommand({ name });
  const createStageResponse = await this.ivsRealtimeClient.send(createStageRequest);
  this.stages[name] = createStageResponse.stage;
  this.stageParticipants[name] = [];
  return this.stages[name];
}; Generating Stage Participant Tokens I created another route in my application to handle generating participant tokens called . /ivs-stage-token router.post('/ivs-stage-token', async function (req, res, next) {
  const body = req.body;
  const username = body.username;
  const stageName = body.stageName;
  const userId = uuid4();
  let token = await ivsRealtimeService.createStageParticipantToken(stageName, userId, username);
  res.json(token);
}); Note that this endpoint requires a , , and a and invokes . username stageName userId ivsRealtimeService.createStageParticipantToken() async createStageParticipantToken(stageName, userId, username, duration = 60) {
  let stage;
  if (!this.stageExists(stageName)) {
    stage = await this.createStage(stageName)
  }
  else {
    stage = this.getStage(stageName);
  }
  const stageArn = stage.arn;

  const createStageTokenRequest = new CreateParticipantTokenCommand({
    attributes: {
      username,
    },
    userId,
    stageArn,
    duration,
  });
  const createStageTokenResponse = await this.ivsRealtimeClient.send(createStageTokenRequest);
  const participantToken = createStageTokenResponse.participantToken;
  this.stageParticipants[stageName].push(participantToken);
  return participantToken;
}; In this method, we're creating a which expects an input object containing the , the , the token duration (default: 60 minutes), and an object which can be used to store arbitrary application-specific values (in my case, the ). CreateParticipantTokenCommand userId stageArn attributes username The will be available later on when we create our front-end, so it's a nice way to include participant-specific information. But, like our docs say: attributes This field is exposed to all stage participants and should not be used for personally identifying, confidential, or sensitive information. Now that we've created a few endpoints to help us create stage resources and participant tokens, let's look at creating the web app. Building the Web Application The front-end will be a straightforward, vanilla JavaScript and HTML application to keep things simple and focus on learning the Web Broadcast SDK. For this demo, we can add a route that returns an HTML file. In that file, include the Web Broadcast SDK (version ). 1.3.2 The Includes and Markup <script src="https://web-broadcast.live-video.net/1.3.2/amazon-ivs-web-broadcast.js"></script> Because the number of participants in this virtual stage is dynamic (up to 12), it makes sense to create a that contains a tag and any other buttons, labels, or markup that we need. Here's how that might look. <template> <video> <template id="stages-guest-template">
  <video class="participant-video" autoplay></video>
  <div>
    <small class="participant-name"></small>
  </div>
  <div>
    <button type="button" class="settings-btn">Cam/Mic Settings</button>
  </div>
</template> I also have an empty that will be used to render the participants as they join the virtual stage. <div> <div id="participants"></div> The JavaScript Now that we have the Web Broadcast SDK dependency and the markup ready to go, we can look at the JavaScript required to join a virtual "stage" and render the participants. This involves several steps, but we can break them out into manageable functions to keep things simple. When the DOM is ready, we can call the following functions (we'll look at each below). let
  audioDevices,
  videoDevices,
  selectedAudioDeviceId,
  selectedVideoDeviceId,
  videoStream,
  audioStream,
  username,
  stageConfig,
  username = '[USERNAME]',
  stageName = '[STAGE NAME]',
  stageParticipantToken,
  stage;

document.addEventListener('DOMContentLoaded', async () => {
  await handlePermissions();
  await getDevices();
  await createVideoStream();
  await createAudioStream();
  stage = await getStageConfig(stageName);
  stageParticipantToken = await getStageParticipantToken(stage.name, username);
  await initStage();
}); Devices and Permissions The first 4 method calls ( , , , ) should look familiar if you've worked with the Amazon IVS Web Broadcast SDK in the past. handlePermissions() getDevices() createVideoStream() createAudioStream() We'll quickly look at each function below, but you can always refer to the for more information. docs First, lets us prompt the user for permission to access their webcam and microphone. handlePermissions() const handlePermissions = async () => {
  let permissions;
  try {
    const stream = await navigator.mediaDevices.getUserMedia({ video: true, audio: true });
    for (const track of stream.getTracks()) {
      track.stop();
    }
    permissions = { video: true, audio: true };
  }
  catch (err) {
    permissions = { video: false, audio: false };
    console.error(err.message);
  }
  if (!permissions.video) {
    console.error('Failed to get video permissions.');
  } else if (!permissions.audio) {
    console.error('Failed to get audio permissions.');
  }
}; Next, retrieves a list of webcams and microphones and stores them. getDevices() In this demo, I'm defaulting the selected video and audio device to the first available device, but in your application, you'd likely be presenting these in a to let the user pick which broadcast device they'd like to use. <select> const getDevices = async () => {
  const devices = await navigator.mediaDevices.enumerateDevices();
  videoDevices = devices.filter((d) => d.kind === 'videoinput');
  audioDevices = devices.filter((d) => d.kind === 'audioinput');
  selectedVideoDeviceId = videoDevices[0].deviceId;
  selectedAudioDeviceId = audioDevices[0].deviceId;
}; Now that we have the devices listed, we can create a video and audio stream from them. For multi-host videos, we should make sure that we keep in mind the for framerate and video size. recommended limits const createVideoStream = async () => {
  videoStream = await navigator.mediaDevices.getUserMedia({
    video: {
      deviceId: {
        exact: selectedVideoDeviceId
      },
      width: {
        ideal: 1280,
        max: 1280,
      },
      height: {
        ideal: 720,
        max: 720,
      },
      frameRate: {
        max: 30,
      },
    },
    });
};

const createAudioStream = async () => {
  audioStream = await navigator.mediaDevices.getUserMedia({
    audio: {
      deviceId: selectedAudioDeviceId
    },
  });
}; Configuring and Joining The Stage Now that we have permissions, devices, and streams sorted we can focus on the virtual stage. First, declare some necessary variables. const { Stage, SubscribeType, LocalStageStream, StageEvents, StreamType } = IVSBroadcastClient; Now we can call the method which calls the API endpoint that we created above to create (or retrieve) the stage resource. getStageConfig() The value of the stage here is how our application will allow multiple participants to join the same virtual stage, so we'd probably want to pass this in somehow (maybe via URL variables or retrieved from a backend). name const getStageConfig = async (name) => {
  const stageRequest = await fetch('/ivs-stage', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ name })
  });
  const stage = await stageRequest.json();
  return stage;
}; Next, retrieves a token from the other API endpoint that we created above. getStageParticipantToken() The variable is a property of the object returned from immediately above, and the depends on your application logic (maybe you have access to a current logged-in user property). stageName stage getStageConfig() username const getStageParticipantToken = async (stageName, username) => {
  const stageTokenRequest = await fetch('/ivs-stage-token', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ username, stageName }),
  });
  const token = await stageTokenRequest.json();
  return token;
}; Now, we're ready to configure and join the stage. The method will create an instance of the object, which expects the stage participant token and a object. initStage() Stage strategy The object contains three functions that express the desired state of the stage (refer to the for all of the different possibilities for strategy). Within the method, we can define it like so. strategy docs initStage() const initStage = async () => {
  const strategy = {
    shouldSubscribeToParticipant: (participant) => {
      return SubscribeType.AUDIO_VIDEO;
    },
    shouldPublishParticipant: (participant) => {
      return true;
    },
    stageStreamsToPublish: () => {
      const videoTrack = videoStream.getVideoTracks()[0]
      const audioTrack = audioStream.getAudioTracks()[0]
      const streamsToPublish = [
        new LocalStageStream(videoTrack),
        new LocalStageStream(audioTrack)
      ];
      return streamsToPublish;
    },
  };
} Before we move forward with the rest of the method, let's break down the strategy object. The first function ( ) expresses how to handle each participant that joins the application. initStage() shouldSubscribeToParticipant() This gives us the freedom to add participants who may act as moderators ( ), audio-only participants ( ), or participants with full video and audio ( ) as shown above. NONE AUDIO AUDIO_VIDEO Next, expresses whether or not the participant should be published. You might want to check the state of a participant based on a button click or a check box to give participants the ability to remain unpublished until they are ready. shouldPublishParticipant() Finally, expresses an array of objects containing the elements that should be published. In this demo, we'll use both the and that we created above to generate these. stageStreamsToPublish() LocalStageStream MediaStream videoStream audioStream Next, inside of the method, we create an instance of the class, passing it the participant token and the strategy. initStage() Stage stage = new Stage(stageParticipantToken.token, strategy); Now that we have a instance, we can attach listeners to the various events that are broadcast on the stage. See the . In this demo, we'll listen for when participants are added or removed. Stage docs for all of the possible events When participants are added, we will render the participant. stage.on(StageEvents.STAGE_PARTICIPANT_STREAMS_ADDED, (participant, streams) => {
  renderParticipant(participant, streams);
}); The method, defined outside of the method, clones the that we defined above and customizes it for the given participant. renderParticipant() initStage() <template> Note that the object contains a boolean value ; we can use this to only add the video stream for the current local participant to avoid echoing their own voice back to them. participant isLocal const renderParticipant = (participant, streams) => {
  // clone the <template>
  const guestTemplate = document.getElementById('stagesGuestTemplate');
  const newGuestEl = guestTemplate.content.cloneNode(true);

  // populate the template values
  newGuestEl.querySelector('.participant-col').setAttribute('data-participant-id', participant.id);
  newGuestEl.querySelector('.participant-name').textContent = participant.attributes.username;

  // get a list of streams to add
  let streamsToDisplay = streams;
  if (participant.isLocal) {
    streamsToDisplay = streams.filter(stream => stream.streamType === StreamType.VIDEO)
  }

  // add all audio/video streams to the <video>
  const videoEl = newGuestEl.querySelector('.participant-video');
  videoEl.setAttribute('id', `${participant.id}-video`);
  const mediaStream = new MediaStream();
  streamsToDisplay.forEach(stream => {
    mediaStream.addTrack(stream.mediaStreamTrack);
  });
  videoEl.srcObject = mediaStream;

  // add the cloned template to the list of participants
  document.getElementById('participants').appendChild(newGuestEl);
}; Back in the , we can listen for when a participant leaves the stage so that we can remove their video from the DOM. initStage() stage.on(StageEvents.STAGE_PARTICIPANT_STREAMS_REMOVED, (participant, streams) => {
  const videoId = `${participant.id}-video`
  document.getElementById(videoId).closest('.participant-col').remove();
}); For this demo, we won't add any additional listeners. Your business needs will dictate additional listeners, and your application can respond to those as necessary. For example, if we wanted to update an indicator on the client side with the current connection state, we could listen for and set the state indicator each time the handler is invoked. StageEvents.STAGE_CONNECTION_STATE_CHANGED The final step inside is to join the stage. initStage() try {
   await stage.join();
} 
catch (error) {
   // handle join exception
} Leaving the Stage It's not completely necessary, but we can improve the user experience by explicitly leaving a stage when the participant exits the application. This will ensure that the remaining participants’ UI will be updated sooner than later. For this, we can use a handler to invoke to ensure a clean disconnection. beforeunload stage.leave() const cleanUp = () => {
  if (stage) stage.leave();
};

document.addEventListener("beforeunload", cleanUp); Now, our application is ready to test. Running the application gives us a real-time video chat experience between up to 12 participants. Summary In this post, we learned how to create a real-time video chat experience for up to 12 participants. In our next post, we'll learn how to take the next step and broadcast a real-time chat to an Amazon IVS channel so that end viewers can watch the conversation with high quality and low latency. Also published here

How to Create a Real Time Multi Host Video Chat in a Browser With Amazon IVS: A Guide

This is the fourth post in a series all about uploading files for the web. In the previous posts, we covered , , and how to . uploading files using just HTML uploading files using JavaScript receive file uploads on a Node.js server This post is going to take a step back and explore architectural changes to reduce costs when adding file uploads to our applications. https://www.youtube.com/watch?v=cJ6IuSJabXk&embedable=true By this point, we should be receiving in Node.js, parsing the request, capturing the file, and writing that file to the disk on the application server. multipart/form-data There are a couple of issues with this approach. First, this approach doesn’t work for distributed systems that rely on several different machines. If a user uploads a file it can be hard (or impossible) to know which machine received the request, and therefore, where the file is saved. This is especially true if you’re using serverless or edge compute. Secondly, storing uploads on the application server can cause the server to quickly run out of disk space. At which point, we’d have to upgrade our server. That could be much more expensive than other cost-effective solutions. And that’s where comes in. Object Storage What is Object Storage You can think of Object Storage like a folder on a computer. You can put any files (aka “objects”) you want in it, but the folders (aka “buckets”) live within a cloud service provider. You can also access files via URL. Object Storage provides a couple of benefits: It’s a single, central place to store and access all of your uploads. It’s designed to be highly available, easily scalable, and super cost-effective. For example, if you consider , you could run an application for $5/month and get 25 GB of disk space. If your server starts running out of space, you could upgrade your server to get an additional 25 GB, but that’s going to cost you $7/month more. shared CPU servers Alternatively, you could put that money towards Object Storage and you would get 250 GB for $5/month. So 10 times more storage space for less cost. Of course, there are other reasons to upgrade your application server. You may need more RAM or CPU, but if we’re talking purely about disk space, Object Storage is a much cheaper solution. With that in mind, the rest of this article will cover connecting an existing Node.js application to an Object Storage provider. We’ll use to parse multipart requests, but configure it to upload files to Object Storage instead of writing to disk. formidable If you want to follow along, you will need to have an Object Storage bucket set up, as well as the access keys. Any S3-compatible Object Storage provider should work. Today, I’ll be using (formerly Linode). If you want to do the same, here’s a guide that shows you how to get going: Akamai’s cloud computing services https://www.linode.com/docs/products/storage/object-storage/get-started/ And here’s a link to get $100 in free credits for 60 days What is S3 Before we start writing code, there’s one more concept that I should explain, S3. S3 stands for “Simple Storage Service”, and it’s an Object Storage product originally developed at AWS. Along with their product, AWS came up with a standard communication protocol for interacting with their Object Storage solution. As more companies started offering Object Storage services, they decided to also adopt the same S3 communication protocol for their Object Storage service, and S3 became a standard. As a result, we have more options to choose from for Object Storage providers and fewer options to dig through for tooling. We can use the same libraries (maintained by AWS) with other providers. That’s great news because it means the code we write today should work across any S3-compatible service. The libraries we’ll use today are and : @aws-sdk/client-s3 @aws-sdk/lib-storage npm install @aws-sdk/client-s3 @aws-sdk/lib-storage These libraries will help us upload objects into our buckets. Okay, let’s write some code! Start with existing Node.js application We’ll start with an example Nuxt.js event handler that writes files to disk using formidable. It checks if a request contains and if so, it passes the underlying Node.js request object (aka ) to a custom function . Since this function uses the Node.js request, it will work in any Node.js environment and tools like formidable. multipart/form-data IncomingMessage parseMultipartNodeRequest import formidable from 'formidable';

/* global defineEventHandler, getRequestHeaders, readBody */

/**
 * @see https://nuxt.com/docs/guide/concepts/server-engine
 * @see https://github.com/unjs/h3
 */
export default defineEventHandler(async (event) => {
  let body;
  const headers = getRequestHeaders(event);

  if (headers['content-type']?.includes('multipart/form-data')) {
    body = await parseMultipartNodeRequest(event.node.req);
  } else {
    body = await readBody(event);
  }
  console.log(body);

  return { ok: true };
});

/**
 * @param {import('http').IncomingMessage} req
 */
function parseMultipartNodeRequest(req) {
  return new Promise((resolve, reject) => {
    const form = formidable({ multiples: true });
    form.parse(req, (error, fields, files) => {
      if (error) {
        reject(error);
        return;
      }
      resolve({ ...fields, ...files });
    });
  });
} We’re going to modify this code to send the files to an S3 bucket instead of writing them to disk. Set up S3 Client The first thing we need to do is set up an S3 Client to make the upload requests for us, so we don’t have to write them manually. We’ll import the constructor from as well as the command from . We’ll also import Node’s module to use later on. S3Client @aws-sdk/client-s3 Upload @aws-sdk/lib-storage stream import stream from 'node:stream';
import { S3Client } from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage'; Next, we need to configure our client using our S3 bucket , , , and . Again, you should already have set up an S3 bucket and know where to find this information. If not, ( ). endpoint access key secret access key region check out this guide $100 credit I like to store this information in environment variables and not hard-code the configuration into the source code. We can access those variables using to use in our application. process.env const { S3_URL, S3_ACCESS_KEY, S3_SECRET_KEY, S3_REGION } = process.env; If you’ve never used environment variables, they’re a good place for us to put secret information such as access credentials. You can read more about them here: https://nodejs.dev/en/learn/how-to-read-environment-variables-from-nodejs/ With our variables set up, I can now instantiate the S3 Client we’ll use to communicate to our bucket. const s3Client = new S3Client({
  endpoint: `https://${S3_URL}`,
  credentials: {
    accessKeyId: S3_ACCESS_KEY,
    secretAccessKey: S3_SECRET_KEY,
  },
  region: S3_REGION,
}); It’s worth pointing out that the endpoint needs to include the HTTPS protocol. In Akamai’s Object Storage dashboard, when you copy the bucket URL, but it doesn’t include the protocol ( ). So I just add the prefix here. bucket-name.bucket-region.linodeobjects.com With our S3 client configured, we can start using it. Modify formidable In our application, we’re passing any multipart Node request into our custom function, . This function returns a Promise and passes the request to formidable, which parses the request, writes files to the disk, and resolves the promise with the form fields data and files data. parseMultipartNodeRequest function parseMultipartNodeRequest(req) {
  return new Promise((resolve, reject) => {
    const form = formidable({ multiples: true });
    form.parse(req, (error, fields, files) => {
      if (error) {
        reject(error);
        return;
      }
      resolve({ ...fields, ...files });
    });
  });
} This is the part that needs to change. Instead of processing the request and writing files to disk, we want to pipe file streams to an S3 upload request. So as each file chunk is received, it’s passed through our handler to the S3 upload. We’ll still return a promise and use formidable to parse the form, but we have to change formidable’s . We’ll set the option to a function called that we’ll write shortly. configuration options fileWriteStreamHandler fileWriteStreamHandler /** @param {import('formidable').File} file */
function fileWriteStreamHandler(file) {
  // TODO
}
const form = formidable({
  multiples: true,
  fileWriteStreamHandler: fileWriteStreamHandler,
}); Here’s what their documentation says about : fileWriteStreamHandler – default , which by default writes to host machine file system every file parsed; The function should return an instance of a that will receive the uploaded file data. With this option, you can have any custom behavior regarding where the uploaded file data will be streamed for. If you are looking to write the file uploaded in other types of cloud storages (AWS S3, Azure blob storage, Google cloud storage) or private file storage, this is the option you’re looking for. When this option is defined the default behavior of writing the file in the host machine file system is lost. options.fileWriteStreamHandler {function} null Writable stream As formidable parses each chunk of data from the request, it will pipe that chunk into the Writable stream that’s returned from this function. So our function is where the magic happens. fileWriteStreamHandler Before we write the code, let’s understand some things: This function must return a to write each upload chunk to. Writable stream It needs to pipe each chunk of data to an S3 Object Storage. also We can use the command from to create the request. Upload @aws-sdk/lib-storage The request body can be a stream, but it must be a , not a Writable stream. Readable stream A can be used as a Readable and Writable stream. Passthrough stream both Each request formidable will parse may contain multiple files, so we may need to track multiple S3 upload requests. receives one parameter of type interface with properties like , , , and more. fileWriteStreamHandler formidable.File originalFilename size mimetype OK, now let’s write the code. We’ll start with an to store and track all the S3 upload request outside the scope of . Array fileWriteStreamHandler Inside , we’ll create the stream that will serve as both the Readable body of the S3 upload and the Writable return value of this function. We’ll create the request using the S3 libraries, and tell it our bucket name, the object key (which can include folders), the object Content-Type, the Access Control Level for this object, and the stream as the request body. We’ll instantiate the request using and add the returned to our tracking . We might want to add the response property to the object when the upload completes, so we can use that information later on. Lastly, we’ll return the stream from this function: fileWriteStreamHandler Passthrough Upload Passthrough Upload.done() Promise Array Location file Passthrough /** @type {Promise<any>[]} */
const s3Uploads = [];

/** @param {import('formidable').File} file */
function fileWriteStreamHandler(file) {
  const body = new stream.PassThrough();
  const upload = new Upload({
    client: s3Client,
    params: {
      Bucket: 'austins-bucket',
      Key: `files/${file.originalFilename}`,
      ContentType: file.mimetype,
      ACL: 'public-read',
      Body: body,
    },
  });
  const uploadRequest = upload.done().then((response) => {
    file.location = response.Location;
  });
  s3Uploads.push(uploadRequest);
  return body;
} A couple of things to note: is the name and location the object will exist. It can include folders that will be created if they do not currently exist. If a file exists with the same name and location, it will be overwritten (fine for me today). You can avoid collisions by using hashed names or timestamps. Key is not required, but it’s helpful to include. It allows browsers to create the downloaded response appropriately based on Content-Type. ContentType : is also optional, but by default, every object is private. If you want people to be able to access the files via URL (like an element), you’ll want to make it public. ACL <img> Although supports uploads, you need to support Readable streams. @aws-sdk/client-s3 @aws-sdk/lib-storage You can read more about the parameters on NPM: https://www.npmjs.com/package/@aws-sdk/client-s3 This way, formidable becomes the plumbing that connects the incoming client request to the S3 upload request. Now there’s just one more change to make. We are keeping track of all the upload requests, but we aren’t waiting for them to finish. We can fix that by modifying the function. It should continue to use formidable to parse the client request, but instead of resolving the promise immediately, we can use to wait until all the upload requests have resolved. parseMultipartNodeRequest Promise.all The whole function looks like this: /**
 * @param {import('http').IncomingMessage} req
 */
function parseMultipartNodeRequest(req) {
  return new Promise((resolve, reject) => {
    /** @type {Promise<any>[]} */
    const s3Uploads = [];

    /** @param {import('formidable').File} file */
    function fileWriteStreamHandler(file) {
      const body = new PassThrough();
      const upload = new Upload({
        client: s3Client,
        params: {
          Bucket: 'austins-bucket',
          Key: `files/${file.originalFilename}`,
          ContentType: file.mimetype,
          ACL: 'public-read',
          Body: body,
        },
      });
      const uploadRequest = upload.done().then((response) => {
        file.location = response.Location;
      });
      s3Uploads.push(uploadRequest);
      return body;
    }
    const form = formidable({
      multiples: true,
      fileWriteStreamHandler: fileWriteStreamHandler,
    });
    form.parse(req, (error, fields, files) => {
      if (error) {
        reject(error);
        return;
      }
      Promise.all(s3Uploads)
        .then(() => {
          resolve({ ...fields, ...files });
        })
        .catch(reject);
    });
  });
} The resolved value will also contain the property we included, pointing to the Object Storage URL. files location Walk through the whole flow We covered a lot, and I think it’s a good idea to review how everything works together. If we look back at the original event handler, we can see that any request will be received and passed to our function. The resolved value from this function will be logged to the console: multipart/form-data parseMultipartNodeRequest export default defineEventHandler(async (event) => {
  let body;
  const headers = getRequestHeaders(event);

  if (headers['content-type']?.includes('multipart/form-data')) {
    body = await parseMultipartNodeRequest(event.node.req);
  } else {
    body = await readBody(event);
  }
  console.log(body);

  return { ok: true };
}); With that in mind, let’s break down what happens if I want to upload . a cute photo of Nugget making a big ol’ yawn For the browser to send the file as binary data, it needs to make a request with an or . multiplart/form-data HTML form with JavaScript Our Nuxt.js and passes the underlying Node.js request object to our custom function. application receives the multipart/form-data parseMultipartNodeRequest returns a that will eventually be resolved with the data. Inside that , we instantiate the formidable library and pass the request object to formidable for parsing. parseMultipartNodeRequest Promise Promise As formidable is parsing the request when it comes across a file, it writes the chunks of data from the file stream to the stream that’s returned from the function. Passthrough fileWriteStreamHandler Inside the we also set up a request to upload the file to our S3-compatible bucket, and we use the same stream as the body of the request. So as formidable writes chunks of file data to the stream, they are also read by the S3 upload request. fileWriteStreamHandler Passthrough Passthrough Once formidable has finished parsing the request, all the chunks of data from the file streams are taken care of, and we wait for the list of S3 requests to finish uploading. After all that is done, we resolve the from with the modified data from formidable. The variable is assigned to the resolved value. Promise parseMultipartNodeRequest body The data representing the fields and files (not the files themselves) are logged to the console. So now, if our original upload request contained a single field called “file1” with the photo of Nugget, we might see something like this: {
  file1: {
    _events: [Object: null prototype] { error: [Function (anonymous)] },
    _eventsCount: 1,
    _maxListeners: undefined,
    lastModifiedDate: null,
    filepath: '/tmp/93374f13c6cab7a01f7cb5100',
    newFilename: '93374f13c6cab7a01f7cb5100',
    originalFilename: 'nugget.jpg',
    mimetype: 'image/jpeg',
    hashAlgorithm: false,
    createFileWriteStream: [Function: fileWriteStreamHandler],
    size: 82298,
    _writeStream: PassThrough {
      _readableState: [ReadableState],
      _events: [Object: null prototype],
      _eventsCount: 6,
      _maxListeners: undefined,
      _writableState: [WritableState],
      allowHalfOpen: true,
      [Symbol(kCapture)]: false,
      [Symbol(kCallback)]: null
    },
    hash: null,
    location: 'https://austins-bucket.us-southeast-1.linodeobjects.com/files/nugget.jpg',
    [Symbol(kCapture)]: false
  }
} It looks very similar to the object formidable returns when it writes directly to disk, but this time it has an extra property, , which is the Object Storage URL for our uploaded file. location Throw that sucker in your browser and what do you get? That’s right! A cute photo of Nugget making a big ol’ yawn 🥰 I can also go to my bucket in my and see that I now have a folder called “files” containing a file called “nugget.jpg”. Object Storage dashboard Closing thoughts Okay, we covered a lot today. I hope it all made sense. If not, feel free to reach out to me with questions. Also, reach out and let me know if you got it working in your own application. I’d love to hear from you because using Object Storage is an excellent architectural decision if you need a single, cost-effective place to store files. In the next posts, we’ll work on making our applications deliver files faster, as well as protecting our applications from malicious uploads. I hope you stick around. Thank you so much for reading. If you liked this article, and want to support me, the best ways to do so are to , , and . share it sign up for my newsletter follow me on Twitter Originally on . published austingil.com

Stream File Uploads to S3 Object Storage and Save Money

I've blogged quite a bit about the Amazon Interactive Video Service (Amazon IVS) Web Broadcast SDK. We've learned about , how to use the SDK to , how to , and even looked at using it to . the basics stream pre-recorded videos add screen sharing and overlays create a Lofi radio station The Web Broadcaster SDK is a game changer that gives developers the ability to integrate the broadcast experience directly into their streaming applications instead of directing their users to use third-party desktop streaming software. In this post, we'll look at another exciting possible use case of the Web Broadcast SDK: streaming browser based games directly to an Amazon IVS channel. It's really quite easy to do, and we'll take it a step further by creating an interactive experience that allows live stream viewers to control the gameplay (akin to ). TwitchPlaysPokemon Streaming a Browser Based Game Directly From the Browser For this post, I'll assume that you're already familiar with Amazon IVS and have already configured a live streaming channel. If you're new to Amazon IVS, check out the blog series , specifically the very . Optionally, you could also refer to the which is a great resource for learning how to develop live streaming applications with Amazon IVS. Getting Started with Amazon IVS first post user guide Since I'm not a game developer, I decided to add live streaming to a few existing open source, browser-based games: and . Since both of these games utilize for gameplay, it will be easy to obtain a from them will be the source for our live stream video input. pacman-canvas Astray <canvas> MediaStream pacman-canvas For the first demo, I cloned the repo to my local machine and took a look at the code. pacman-canvas Initializing the Broadcast Client The game uses jQuery, so I added a call to an method to the end of the existing DOM ready handler: pacman-canvas initBroadcast() let broadcastClient;
let isBroadcasting = false;

$(document).ready(function () {
  // game logic...
  initBroadcast();
}) In my method, I create an instance of the ( ), passing it the from my Amazon IVS channel. initBroadcast() AmazonIVSBroadcastClient docs Ingest endpoint broadcastClient = IVSBroadcastClient.create({
  streamConfig: IVSBroadcastClient.STANDARD_LANDSCAPE,
  ingestEndpoint: config.ingestEndpoint,
}); Next, to add the gameplay to the client, I grabbed a reference to the element used by the game and called ( ) on the : <canvas> addVideoInputDevice() docs broadcastClient const game = document.getElementById('myCanvas');
broadcastClient.addVideoInputDevice(game.captureStream(), 'game-track', { index: 0 }); Finally, to start the broadcast, I call and pass it the from my channel. startBroadcast() Stream key broadcastClient
  .startBroadcast(config.streamKey)
  .then(() => {
    isBroadcasting = true;
    console.log('online')
  })
  .catch((error) => {
    isBroadcasting = false;
    console.error(error);
  }); The game uses a few tags for sound effects and calls the following function as necessary. <audio> var Sound = {};
Sound.play = function (sound) {
  if (game.soundfx == 1) {
    var audio = document.getElementById(sound);
    (audio !== null) ? audio.play() : console.log(sound + " not found");
  }
}; To add the game audio to my stream, I modified it to call ( ). addAudioInputDevice() docs var Sound = {};
Sound.play = function (sound) {
  if (game.soundfx == 1) {
    var audio = document.getElementById(sound);
    (audio !== null) ? audio.play() : console.log(sound + " not found");
    var trackLabel = `${sound}-audio-track-${new Date().getTime()}`;
    audio.addEventListener('playing', (evt) => {
      if (!broadcastClient.getAudioInputDevice(trackLabel)) {
        broadcastClient.addAudioInputDevice(audio.captureStream(), trackLabel);
      }
    });
  }
}; And that's it! The gameplay stream will be broadcast whenever the method is called, so we could attach that to a button click handler or call it from the method of the existing game. If we wanted to, we could also and even size and position the webcam as an overlay (see ( ) for more info). startBroadcast() newGame add a webcam to the stream VideoComposition docs Playing Back the Live Stream in a Browser For playback, I added a simple HTML page that includes and utilizes the . Amazon IVS Player SDK <script src="https://player.live-video.net/1.17.0/amazon-ivs-player.min.js"></script>
<script>
document.addEventListener('DOMContentLoaded', () => {
  const videoPlayer = document.getElementById('video-player');
  const streamUrl = '[CHANNEL PLAYBACK URL]';
  const ivsPlayer = IVSPlayer.create(); 
  ivsPlayer.attachHTMLVideoElement(videoPlayer);
  ivsPlayer.load(streamUrl);
  ivsPlayer.play();
});
</script>
<body>
  <video id="video-player" />
</body> Here's how everything looks at this point. On the left side is the broadcast/gamer view, and on the right is the playback/viewer view. Adding Timed Metadata As you can see in the Gif above, the score, level, and lives data is not included in the live stream. This is because the game developer did not include those elements in the gameplay . If we wanted to, we could render these as overlays in the playback view by . In my demo, I modified the function of to publish the score as timed metadata on my live stream by calling an AWS Lambda function that I've previously created for this channel. <canvas> publishing them with timed metadata Score pacman-canvas const publishMetadata = async (meta) => {
  await fetch('[lambda url]/send-metadata', {
    method: 'POST',
    headers: {
      'Accept': 'application/json',
      'Content-Type': 'application/json'
    },
    body: JSON.stringify(meta),
  });
};

function Score() {
  this.score = 0;
  this.set = function (i) {
    this.score = i;
  };
  this.add = function (i) {
    this.score += i;
    publishMetadata({
      metadata: JSON.stringify({ score: this.score }),
    });
  };
  this.refresh = function (h) {
    $(h).html("Score: " + this.score);
  };

} Be careful publishing metadata too frequently to avoid exceeding the Amazon IVS . You may need to throttle or batch metadata publishing to avoid exceeding the quota. Note: service quotas Adding Interactivity to a Browser Based Game Live Stream Now let's take a look at how we might create an interactive browser based game that can be controlled by live stream viewers. For this, I chose - a simple game that requires the player to navigate a ball through a maze. I added the Web Broadcast client just as i did with above, and decided to use Amazon IVS chat to receive the gameplay control commands from the viewer. Astray pacman-canvas Adding Chat We've previously looked at how to . To use chat for game interactivity, we'll configure a chat experience just like we usually do, but in the chat connection's message handler, we'll check the message content for the terms , , and to control the ball movement on the broadcaster side. add chat to your live streaming application with Amazon IVS chat left right up down const chatConnection = new WebSocket('wss://edge.ivschat.us-east-1.amazonaws.com', token);

chatConnection.onmessage = (event) => {
  const data = JSON.parse(event.data);
  const chatEl = document.getElementById('chat');
  if (data.Type === 'MESSAGE') {
    const direction = data.Content.toLowerCase();
    if (['left', 'up', 'right', 'down'].indexOf(direction) > -1) moveBall(direction);
    // render message to chat div
  }
} If I find one of the directional commands in the incoming message, I call which will simulate a key press on the broadcaster side to move the ball in the game. moveBall() const moveBall = (direction) => {
  let keyCode;
  let keyCodes = {
    left: 37,
    up: 38,
    right: 39,
    down: 40,
  }
  const keyDown = new KeyboardEvent('keydown', { keyCode: keyCodes[direction] });
  const keyUp = new KeyboardEvent('keyup', { keyCode: keyCodes[direction] });
  document.dispatchEvent(keyDown);
  setTimeout(() => {
    document.dispatchEvent(keyUp);
  }, 200);
}; There's a minor catch to this approach. Because uses , the broadcaster's game tab must be active/visible tab at all times since to improve performance and battery life. Note: Astray requestAnimationFrame() requestAnimationFrame() pauses when the tab is not visible Here's how looks when controlled by the viewer. Astray There's a slight delay due to the stream latency, but the potential to give viewers control of the live stream gameplay is intriguing and filled with potential. Summary In this post, we learned how to integrate the Amazon IVS Web Broadcast SDK into a browser based game to give players the ability to live stream gameplay directly to an Amazon IVS live streaming channel. We also learned how to add interactivity to the experience to give stream viewers the ability to directly affect the gameplay. If you'd like to learn more about Amazon IVS, check out the blog post series here on HackerNoon. Getting Started with Amazon IVS Also published . here

Stories, News & Mentions about STREAMKAP

Streamkap’s Stories on HackerNoon