import { check } from "k6";
import {
    writer,
    produceWithConfiguration,
    createTopic,
} from "k6/x/kafka"; 
import http from 'k6/http';
import encoding from 'k6/encoding';

const bootstrapServers = ["xxx:9092"];
const topic = "payin";

const auth = JSON.stringify({
    username: "xxx",
    password: "xxx",
    algorithm: "SHA512",
});

const producer = writer(bootstrapServers, topic, auth);

var configuration = JSON.stringify({
    producer: {
        keySerializer: "io.confluent.kafka.serializers.KafkaAvroSerializer",
        valueSerializer: "io.confluent.kafka.serializers.KafkaAvroSerializer",
    },
    schemaRegistry: {
        url: "xxxx",
        basicAuth: {
            credentialsSource: "USER_INFO",
            userInfo: "xxx:xxx",
        },
    },
});

export default function () {

    const credentials = `xxx:xxx`;
    const encodedCredentials = encoding.b64encode(credentials);
    const options = {
        headers: {
        Authorization: `Basic ${encodedCredentials}`,
        },
    };

const result = http.get('xxxx', options)
const schema = JSON.parse(result.body).schema

let messages = [
    {
        value: JSON.stringify({
            "status": "PROPOSED"
        }),
    },
];

let error = produceWithConfiguration(
    producer,
    messages,
    configuration,
    null,
    schema
);

check(error, {
    "is sent": (err) => err == undefined,
});

}