flutter pc端使用grpc双向流

官网

grpc-dart：https://github.com/grpc/grpc-dart

proto文件

syntax = "proto3";

option go_package = "./";

package helloworld;

service RouteGuide {
  
  rpc GetFeature(Point) returns (Feature) {}

 
  rpc ListFeatures(Rectangle) returns (stream Feature) {}

  
  rpc RecordRoute(stream Point) returns (RouteSummary) {}

  
  rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}
}


message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

message Rectangle {

  Point lo = 1;


  Point hi = 2;
}


message Feature {

  string name = 1;


  Point location = 2;
}

message RouteNote {
.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}


message RouteSummary {

  int32 point_count = 1;

  int32 feature_count = 2;


  int32 distance = 3;


  int32 elapsed_time = 4;
}

通过proto文件生成dart文件

命令
注意：笔者是先将protos文件夹放到根目录，再执行的这条命令
proto文件也是放在protos文件夹下面的
然后将操作后的peotos文件放到lib文件夹下

protoc --dart_out=grpc:protos -Iprotos protos/route_guide.proto

结果如下：
在这里插入图片描述

客户端实现

main.dart

import 'package:flutter/material.dart';
import 'package:app1/client.dart';

void main() {
  runApp(const MyApp());
}

class MyApp extends StatelessWidget {
  const MyApp({super.key});

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      debugShowCheckedModeBanner: false, //去掉debug的标志
      title: "fltter Demo",
      theme: ThemeData(primaryColor: Color.fromRGBO(0, 137, 255, 1)),
      home: const HomePage(),
    );
  }
}

class HomePage extends StatefulWidget {
  const HomePage({super.key});

  @override
  State<HomePage> createState() => _HomePageState();
}

class _HomePageState extends State<HomePage> {
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      body: Center(
          child: Column(
        mainAxisAlignment: MainAxisAlignment.center,
        children: [
          ElevatedButton(
            onPressed: () {
              () async {
                print("发送");
                List<String> args = [];
                await Client().main(args);
              }();
            },
            child: const Text("发送消息"),
          ),
        ],
      )),
    );
  }
}

common.dart

import 'dart:convert';
import 'dart:io';

import './protos/route_guide.pb.dart';

const coordFactor = 1e7;

final List<Feature> featuresDb = _readDatabase();

List<Feature> _readDatabase() {
  final dbData = File('data/route_guide_db.json').readAsStringSync();
  final List db = jsonDecode(dbData);
  return db.map((entry) {
    final location = Point()
      ..latitude = entry['location']['latitude']
      ..longitude = entry['location']['longitude'];
    return Feature()
      ..name = entry['name']
      ..location = location;
  }).toList();
}

client.dart

// Copyright (c) 2017, the gRPC project authors. Please see the AUTHORS file
// for details. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

import 'dart:math' show Random;

import 'package:grpc/grpc.dart';

import 'common.dart';
import 'protos/route_guide.pbgrpc.dart';

class Client {
  late RouteGuideClient stub;

  Future<void> main(List<String> args) async {
    final channel = ClientChannel('127.0.0.1',
        port: 8080,
        options:
            const ChannelOptions(credentials: ChannelCredentials.insecure()));
    stub = RouteGuideClient(channel,
        options: CallOptions(timeout: Duration(seconds: 30)));
    // Run all of the demos in order.
    try {
      await runRouteChat();
    } catch (e) {
      print('Caught error: $e');
    }
    await channel.shutdown();
  }

  void printFeature(Feature feature) {
    final latitude = feature.location.latitude;
    final longitude = feature.location.longitude;
    final name = feature.name.isEmpty
        ? 'no feature'
        : 'feature called "${feature.name}"';
    print(
        'Found $name at ${latitude / coordFactor}, ${longitude / coordFactor}');
  }

  /// Run the getFeature demo. Calls getFeature with a point known to have a
  /// feature and a point known not to have a feature.
  Future<void> runGetFeature() async {
    final point1 = Point()
      ..latitude = 409146138
      ..longitude = -746188906;
    final point2 = Point()
      ..latitude = 0
      ..longitude = 0;

    printFeature(await stub.getFeature(point1));
    printFeature(await stub.getFeature(point2));
  }

  /// Run the listFeatures demo. Calls listFeatures with a rectangle containing
  /// all of the features in the pre-generated database. Prints each response as
  /// it comes in.
  Future<void> runListFeatures() async {
    final lo = Point()
      ..latitude = 400000000
      ..longitude = -750000000;
    final hi = Point()
      ..latitude = 420000000
      ..longitude = -730000000;
    final rect = Rectangle()
      ..lo = lo
      ..hi = hi;

    print('Looking for features between 40, -75 and 42, -73');
    await for (var feature in stub.listFeatures(rect)) {
      printFeature(feature);
    }
  }

  /// Run the recordRoute demo. Sends several randomly chosen points from the
  /// pre-generated feature database with a variable delay in between. Prints
  /// the statistics when they are sent from the server.
  Future<void> runRecordRoute() async {
    Stream<Point> generateRoute(int count) async* {
      final random = Random();

      for (var i = 0; i < count; i++) {
        final point = featuresDb[random.nextInt(featuresDb.length)].location;
        print(
            'Visiting point ${point.latitude / coordFactor}, ${point.longitude / coordFactor}');
        yield point;
        await Future.delayed(Duration(milliseconds: 200 + random.nextInt(100)));
      }
    }

    final summary = await stub.recordRoute(generateRoute(10));
    print('Finished trip with ${summary.pointCount} points');
    print('Passed ${summary.featureCount} features');
    print('Travelled ${summary.distance} meters');
    print('It took ${summary.elapsedTime} seconds');
  }

  /// Run the routeChat demo. Send some chat messages, and print any chat
  /// messages that are sent from the server.
  Future<void> runRouteChat() async {
    RouteNote createNote(String message, int latitude, int longitude) {
      final location = Point()
        ..latitude = latitude
        ..longitude = longitude;
      return RouteNote()
        ..message = message
        ..location = location;
    }

    final notes = <RouteNote>[
      createNote('First message', 0, 0),
      createNote('Second message', 0, 1),
      createNote('Third message', 1, 0),
      createNote('Fourth message', 0, 0),
    ];

    Stream<RouteNote> outgoingNotes() async* {
      for (final note in notes) {
        // Short delay to simulate some other interaction.
        await Future.delayed(Duration(milliseconds: 10));
        print('Sending message ${note.message} at ${note.location.latitude}, '
            '${note.location.longitude}');
        yield note;
      }
    }

    final call = stub.routeChat(outgoingNotes());
    await for (var note in call) {
      print(
          'Got message ${note.message} at ${note.location.latitude}, ${note.location.longitude}');
    }
  }
}

服务端

服务端笔者使用的是go来开发的，所以只列出go语言的代码

proto文件

proto文件一定要与上文中提到的一致，否则会报错，错误码为code=12
这里不再赘述

通过proto文件生成go文件命令

注意：要将终端的路径切换到proto文件存在的文件目录下

在这里插入图片描述

protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative route_guide.proto

server.go

package main

import (
	"fmt"
	"google.golang.org/grpc"
	pb "grpc_server/pbfiles"
	"log"
	"net"
	"sync"
	"time"
)

const (
	port = ":8080"
)

// 服务对象
type server struct {
	pb.UnimplementedRouteGuideServer
}

func (s *server) RouteChat(allStr pb.RouteGuide_RouteChatServer) error {
	fmt.Printf("1111")
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		for {
			data, _ := allStr.Recv()
			log.Println(data)
		}
	}()

	go func() {
		for {
			allStr.Send(&pb.RouteNote{
				Location: &pb.Point{
					Latitude:  1,
					Longitude: 1,
				},
				Message: "",
			})
			time.Sleep((time.Second))
		}
	}()

	wg.Wait()
	return nil
}

func main() {
	lis, err := net.Listen("tcp", port)
	if err != nil {
		return
	}
	// 创建一个grpc 服务器
	s := grpc.NewServer()
	// 注册事件
	pb.RegisterRouteGuideServer(s, &server{})
	// 处理链接
	err = s.Serve(lis)
	if err != nil {
		return
	}
}