srctree

alloc: Allocator,

peer: std.net.Stream,

poller: Poller,

//reader: AnyReader,

writer: std.net.Stream.Writer,

drop: usize = 0,

 

pub const Client = @This();

 

pub const Poller = std.io.Poller(PEnum);

pub const PEnum = enum { srv };

 

pub fn init(a: Allocator, host: []const u8, port: u16) !Client {

const peer = try std.net.tcpConnectToHost(a, host, port);

const poller = std.io.poll(

a,

PEnum,

.{ .srv = .{ .handle = peer.handle } },

);

const c: Client = .{

.alloc = a,

.peer = peer,

.poller = poller,

.writer = peer.writer(),

};

return c;

}

 

/// might be rolled into init

pub fn connect(c: *Client) !void {

var w = c.writer.any();

try (Connect{}).send(&w);

}

 

pub fn recv(c: *Client) !Packet.Parsed {

var fifo = c.poller.fifo(.srv);

var ready = fifo.readableLength();

if (c.drop > 0 and ready >= c.drop) {

fifo.discard(c.drop);

c.drop = 0;

}

var poll_more = try c.poller.poll();

while (poll_more) {

ready = fifo.readableLength();

log.err("loop", .{});

 

if (ready < 6) {

poll_more = try c.poller.poll();

continue;

}

 

const pkt: Packet.FixedHeader = @bitCast(fifo.readItem().?);

var fr = fifo.reader();

var r = fr.any();

const reported = try Packet.unpackVarInt(&r);

while (ready < reported and poll_more) {

log.err(" getting more data... {}/{}", .{ ready, reported });

poll_more = try c.poller.poll();

ready = fifo.readableLength();

}

const payload = fifo.readableSliceOfLen(reported);

 

c.drop = reported;

return try Packet.parse(pkt, payload);

}

return error.StreamCrashed;

}

 

const Packet = @import("Packet.zig");

const Publish = @import("Publish.zig");

const Connect = @import("Connect.zig");

const Subscribe = @import("Subscribe.zig");

 

const std = @import("std");

const Allocator = std.mem.Allocator;

const log = std.log.scoped(.mqtt);

const AnyReader = std.io.AnyReader;

const AnyWriter = std.io.AnyWriter;

pub fn parse(r: *AnyReader) !Connect {

_ = r;

@panic("not implemented");

}

 

 

pub fn parse(pkt: []const u8) !Ack {

log.err("Connect Ack not implemented", .{});

log.err("Connect Ack ({}) {any}", .{ pkt.len, pkt });

return .{};

}

const AnyReader = std.io.AnyReader;

pub const Flags = packed struct(u4) {

// TODO find better name

pub const Header = struct {

header: FixedHeader,

length: usize,

};

 

// TODO find better name

pub const Parsed = union(ControlType) {

reserved: void,

CONNECT: Connect,

CONNACK: Connect.Ack,

PUBLISH: Publish,

PUBACK: Publish.Ack,

PUBREC: void,

PUBREL: void,

PUBCOMP: void,

SUBSCRIBE: Subscribe,

SUBACK: Subscribe.Ack,

UNSUBSCRIBE: void,

UNSUBACK: void,

PINGREQ: void,

PINGRESP: void,

DISCONNECT: void,

AUTH: void,

};

 

pub fn parse(header: FixedHeader, payload: []const u8) !Parsed {

var fbs = std.io.fixedBufferStream(payload);

var fbsr = fbs.reader();

var r = fbsr.any();

switch (header.kind) {

.CONNECT => return .{ .CONNECT = try Connect.parse(&r) },

.CONNACK => return .{ .CONNACK = try Connect.Ack.parse(payload) },

.PUBLISH => return .{ .PUBLISH = try Publish.parse(payload, header.flags) },

.PUBACK => return .{ .PUBACK = try Publish.Ack.parse(&r) },

.SUBSCRIBE => return .{ .SUBSCRIBE = try Subscribe.parse(&r) },

.SUBACK => return .{ .SUBACK = try Subscribe.Ack.parse(&r) },

.PUBREC, .PUBREL, .PUBCOMP, .UNSUBSCRIBE, .UNSUBACK, .PINGREQ, .PINGRESP, .DISCONNECT, .AUTH => {

log.err("not implemented parser for {}", .{header.kind});

@panic("not implemented");

},

else => |els| {

log.err("not implemented parser for {}", .{els});

unreachable;

},

}

unreachable;

}

 

const Publish = @import("Publish.zig");

const Connect = @import("Connect.zig");

const Subscribe = @import("Subscribe.zig");

 

pub fn parse(publ: []const u8, flags: Packet.ControlType.Flags) !Publish {

var fbs = std.io.fixedBufferStream(publ);

var r = fbs.reader();

log.err("{s} <> {any}", .{ publ, publ });

const slen = try r.readInt(u16, .big);

const topic = publ[2..][0..slen];

try fbs.seekBy(slen);

var pktid: ?u16 = null;

switch (flags.qos) {

.at_most_once => {

log.err(" expecting {s}", .{"nop"});

},

.at_least_once => {

log.err(" expecting {s}", .{"PUBACK"});

pktid = try r.readInt(u16, .big);

//try Publish.Ack.send(pktid.?, .success, &any);

},

.exactly_once => {

log.err(" expecting {s}", .{"PUBREC"});

},

.invalid => @panic("unreachable"),

}

const props = publ[2 + slen .. 2 + slen]; // TODO implement

try fbs.seekBy(1); // short VLI

 

return .{

.topic_name = topic,

.packet_ident = pktid,

.properties = props,

.payload = publ[fbs.pos..],

};

}

 

pub fn parse(r: *AnyReader) !Ack {

_ = r;

@panic("not implemented");

}

 

const AnyReader = std.io.AnyReader;

pub fn parse(r: *AnyReader) !Subscribe {

_ = r;

@panic("not implemented");

}

 

pub const Ack = struct {

pub fn parse(r: *AnyReader) !Ack {

_ = r;

return .{};

}

};

 

const AnyReader = std.io.AnyReader;

pub const Client = @import("Client.zig");