#Please help. I Need help for kafka microservice header transfer

20 messages · Page 1 of 1 (latest)

undone crypt
#

I was developed a microservice with kafka and nestjs but the problem is when I want to transfer header gateway to service I am unable to do this I have an option for message section but I don't want to do this I want send it kafka header how can I do this

here is my code

export class UserService {
  constructor(
    @Inject('AUTH_MICROSERVICE') private readonly authClient: ClientKafka,
  ) {}
  private readonly detector = new DeviceDetector({
    clientIndexes: true,
    deviceIndexes: true,
    deviceAliasCode: false,
  });
  async subscribeResponse(messages: string[]) {
    for (let i = 0; i < messages.length; i++) {
      this.authClient.subscribeToResponseOf(messages[i]);
    }
    await this.authClient.connect();
  }

  async disconnect() {
    await this.authClient.close();
  }

send(message: string, data: any) {
    return new Promise((resolve, reject) => {
      this.authClient
        .send(message,  JSON.stringify(data))
        .subscribe({
          next: (response) => {
            resolve(response);
          },
          error: (error) => {
            reject(error);
          },
        });
    });
  }

  async profile(profileDto: ProfileDto) {
    return await this.send(profile, profileDto);  // want to send header with this function from gateway
  }

Please suggest way better way if you can suggest me

limpid storm
#

I am sorry, I have really no idea what you're trying to accomplish

undone crypt
#

I am trying to send header when I will send request from gateway

example: let's say I have a gateway and a service name is auth-service(With Kafka) now I want to get profile by gateway and want send jwt token in header so that I can get it in kafka context in auth-service

#

@limpid storm if you have still confusion I will explain more did you understand?

limpid storm
#

If I understand correctly, you want to send headers in a kafka message along with some payload?

In that case, you can send an object with { headers, value } to the kafka client

#

Btw, a but unrelated, but using Kafka in the request/reaponse pattern as an auth service is pretty much exactly what Kafka is not designed for.
Kafka is made for durable, time-ordered log of messages at high volumes. I am pretty sure you don't want your auth data persisted

#

A basic HTTP call to your auth service would make way more sense and would be much simpler to implement.

undone crypt
#

yes you understand correctly but your solution is not work unfortunately. actually I have send some header information with kafka request and also jwt token for security

limpid storm
#

What do you mean "does not work". I am not at the computer currently, so I can't check. But you can definitely send headers with the kafka client

undone crypt
#

is my code is ok?

limpid storm
#

No, value goes into the same object as headers

undone crypt
#

I just put hard value

#

but thanks it's works I checked now

#

thanks a lot

undone crypt
#

@limpid storm please one more thing when I am sending header as nested object where I am wrong this porblem only occured when I giving nested object

it show me this message "The "string" argument must be of type string or an instance of Buffer or ArrayBuffer. Received type number (123)"

how I sending
code:

    return new Promise((resolve, reject) => {
      this.authClient
        .send(message, {
          headers: headers,
          value: JSON.stringify(data),
        })
        .subscribe({
          next: (response) => {
            resolve(response);
          },
          error: (error) => {
            reject(error);
          },
        });
    });
  }


async passwordVerify(passwordVerifyDto: PasswordVerifyDto) {
    return await this.send(
      PASSWORD_VERIFY,
      passwordVerifyDto,
      Buffer.from(
        JSON.stringify({
          token: {
            accessToken: 'This is my access token',
            refreshToken: 'This is my refresh token',
          },
        }),
      ),
    );
  }
limpid storm
#

Headers must be an object with string keys and string or buffer values

undone crypt
#

passwordVerify you can see I convert it buffer

limpid storm